kafka 生产者在消息之间能存活多久？

Question

我正在使用配置属性打开一个 kafka 生产者 -

KafkaProducer<String, MyValue>   producer = new KafkaProducer<String, MyValue>(kafkaProperties);

然后使用-同步发送记录（这样可以避免批处理，也可以保持原来的消息顺序）

   //create myValue instance //omited for simplicity
   //create myrecord instance using topicname and myvalue
   producer.send(myRecord).get();
   producer.flush();  //send message as soon as record is available to producer

现在我的问题是，我有几条记录要发送，并且在发送之间我可能需要等待很长时间 - 几分钟到几小时（无论出于何种原因，至少要更好地探索和理解 kafka）。

我想知道生产者与集群/引导服务器的连接会持续多久。无论如何我可以使用生产者配置来配置它。
（非常感谢深入的解释 - 即使它必须进入 tcp 连接级别，也欢迎您）

（kafka消费者有心跳的概念。生产者有类似的概念吗。谷歌搜索“kafka生产者heartbeat.interval.ms”只返回消费者的结果）。

Answer 1

KafkaProducer.send 方法是异步的，默认情况下它将所有记录添加到缓冲内存中并立即发送，因此根据文档生产者在将批处理发送到集群时建立连接

send() 方法是异步的。当被调用时，它将记录添加到待处理记录发送的缓冲区并立即返回。这允许生产者将单个记录批处理在一起以提高效率。

生产者为每个分区维护未发送记录的缓冲区。这些缓冲区的大小由 batch.size 配置指定。将其增大会导致更多的批处理，但需要更多内存（因为我们通常会为每个活动分区使用这些缓冲区之一）。

默认情况下，即使缓冲区中有额外的未使用空间，缓冲区也可以立即发送。但是，如果您想减少请求的数量，您可以将 linger.ms 设置为大于 0 的值。

这将指示生产者在发送请求之前等待该毫秒数，希望更多记录到达以填满同一批次。这类似于 TCP 中的 Nagle 算法。

例如，在上面的代码 sn-p 中，很可能所有 100 条记录都将在一个请求中发送，因为我们将逗留时间设置为 1 毫秒。但是，如果我们没有填满缓冲区，此设置会为我们的请求增加 1 毫秒的延迟等待更多记录到达。

请注意，即使 linger.ms=0，及时到达的记录通常也会一起批处理，因此无论 linger 配置如何，在重负载下都会发生批处理；但是，将其设置为大于 0 的值可能会导致在不处于最大负载下时请求更少、更高效，但会以少量延迟为代价。

从KafkaProducer.flush 开始，调用flush 并不意味着生产者将每条记录发送到集群，调用flush 会使所有缓冲的记录立即可供发送

调用此方法可以立即发送所有缓冲的记录（即使 linger.ms 大于 0）并阻止与这些记录关联的请求的完成。 flush() 的后置条件是任何先前发送的记录都将完成（例如 Future.isDone() == true）。根据您指定的acks配置成功确认请求时，请求被认为已完成，否则会导致错误。