在Kafka中,可以使用生产者的 send
方法来发送消息,可以批量发送消息的方式有以下几种:
send
方法时,将多条消息封装成一个 ProducerRecord
的列表,然后使用 send
方法一次性发送。示例代码如下:List<ProducerRecord<String, String>> records = new ArrayList<>();
records.add(new ProducerRecord<>("topic", "key1", "value1"));
records.add(new ProducerRecord<>("topic", "key2", "value2"));
records.add(new ProducerRecord<>("topic", "key3", "value3"));
for (ProducerRecord<String, String> record : records) {
producer.send(record).get(); // 同步发送
}
send
方法时,同样将多条消息封装成一个 ProducerRecord
的列表,然后使用 send
方法一次性发送。示例代码如下:List<ProducerRecord<String, String>> records = new ArrayList<>();
records.add(new ProducerRecord<>("topic", "key1", "value1"));
records.add(new ProducerRecord<>("topic", "key2", "value2"));
records.add(new ProducerRecord<>("topic", "key3", "value3"));
for (ProducerRecord<String, String> record : records) {
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
exception.printStackTrace();
} else {
System.out.println("Message sent successfully: " + metadata);
}
}
});
}
以上是两种常见的批量发送消息的方式,可以根据实际需求选择合适的方式。