java - 使用 Kafka 让消费者保持活力

Question

我正在寻找一种“低成本”的方法来在我不主动进行轮询时让消费者保持活力。即，仍在处理上次投票的记录，我不希望消费者连接超时。

一些看起来很有前途的功能:

• wakeup

• commitAsync

在每种情况下，这都是 API 的非标准用法，所以我不确定这是否是一种合理/合理的方法。

RE:将连接超时设置得更高 - 我希望消费者在被阻塞时超时。我的问题与一个部分有关，我在该部分中获取了一个记录 block ，并且单独的线程正在处理它们。

Answer 1

documentation似乎建议您调用 pause() 然后继续主动轮询。如果您在暂停时调用 poll()，则不会返回任何内容。

For use cases where message processing time varies unpredictably, neither of these options may be sufficient. The recommended way to handle these cases is to move message processing to another thread, which allows the consumer to continue calling poll while the processor is still working. Some care must be taken to ensure that committed offsets do not get ahead of the actual position. Typically, you must disable automatic commits and manually commit processed offsets for records only after the thread has finished handling them (depending on the delivery semantics you need). Note also that you will need to pause the partition so that no new records are received from poll until after thread has finished handling those previously returned.

pause() 的文档证实了这一点:

Suspend fetching from the requested partitions. Future calls to poll(long) will not return any records from these partitions until they have been resumed using resume(Collection). Note that this method does not affect partition subscription. In particular, it does not cause a group rebalance when automatic assignment is used.