在kafka中发送同步消息？

Question

如何在 kafka 中发送同步消息？
实现它的一种方法是设置属性参数
max.in.flight.requests.per.connection = 1。

但我想知道是否有在 kafka 中发送同步消息的直接或替代方式。 （类似于 producer.syncSend(...) 等）。

Answer 1

当 max.in.flight.requests.per.connection = 1 时，仅表示在分区内保证消息的顺序与同步无关。

Python 代码以防万一。 对于同步发送，请确保在未来有一个良好的超时时间。

from kafka import KafkaProducer
from kafka.errors import KafkaError

#by default ack = 1, if ack = 'all' --> waits for acks from replicas 
producer = KafkaProducer(bootstrap_servers=['brokerIP:9092'], ack= 'all')


key = b'key'
value = b'value'

future = producer.send("my-topic", key=key, value=value)

# block on this future for sync sends
try:
    record_metadata = future.get(timeout=10)
except KafkaError:
    log.exception()
    pass

print (record_metadata.topic)
print (record_metadata.partition)
print (record_metadata.offset)

producer.flush()
producer.close()

Answer 2

根据我对 Kafka 的冒险 :-) 只有当您有一个 Producer 线程并设置 max.in.flight.requests.per.connection = 1（或转为 retries，即 retries= 0 或两者兼有）时，才能保证消息生产的顺序。

如果您要扩展到多个 Producer，那么您必须“确保”将存储到同一分区的消息将由同一 Producer 实例生成。

Answer 3

正如 Thilo 建议的那样，您可以致电 Future#get 阻止直到发送完成。但是，您可能会遇到一些性能问题，因为当生产者队列有 batch.size 元素时，当大小为 buffer.memory 的缓冲区已满或在 max.block.ms 毫秒之后，生产者开始发送。

如果推送到 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 4

Thilo 提出的答案是可行的方法。一般来说，您关于使用 max.in.flight.requests.per.connection = 1 的建议用于启用仍然重试但不会丢失消息排序。它不那么用于拥有同步生产者。

Answer 5

生产者 API 从 send 返回一个 Future。您可以致电Future#get 阻止直到发送完成。

看到这个example from the 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();