用Kafka发送同步消息？

Question

如何在Kafka中发送同步消息
实现这一点的一种方法是设置properties参数
max.in。航班请求。每连接=1。

但是我想知道是否有一种甚至直接或替代的方式在Kafka中发送同步消息。（比如producer.sync发送（...）等等）。

Answer 1

蒂洛提出的答案是正确的。总的来说，你建议使用max.In。航班请求。每connection=1用于启用仍然重试，但不会丢失消息顺序。它不适合有同步制作人。

Answer 2

正如Thilo所建议的，您可以调用Future#get来阻止，直到发送完成。然而，您可能会遇到一些性能问题，因为当生产者队列中有batch.size元素时，当大小为buffer.memory的缓冲区已满或在<--plhd--2之后，生产者开始发送/>毫秒。

如果有有限的线程推送到kafka，则必须等待max.block。ms每次发送您的信息。因此，在某些情况下，您更喜欢使用：

// send message to producer queue
Future<RecordMetadata> future = producer.send(new ProducerRecord<>(topic, key, message));
// flush producer queue to spare queuing time
producer.flush();
// throw error when kafka is unreachable
future.get(10, TimeUnit.SECONDS);

Answer 3

生产者API从发送返回一个Future。您可以调用Future#get来阻止，直到发送完成。

参见Javadocs中的示例：

如果要模拟简单的阻塞调用，可以立即调用get（）方法：

 byte[] key = "key".getBytes();
 byte[] value = "value".getBytes();
 ProducerRecord<byte[],byte[]> record = 
     new ProducerRecord<byte[],byte[]>("my-topic", key, value)
 producer.send(record).get();