Kafka消息丢失是因为后来的消息

Question

所以我和我的Kafka消费者之间有了一些恼人的矛盾。我使用“Kafka节点”为我的项目。我创造了一个话题。在一个使用者组中通过2台服务器创建了2个使用者。自动提交设置为false。对于我的消费者获得的每一个mesaage，他们会启动一个异步进程，该进程可能需要1~20秒，当进程完成时，消费者会提交偏移量。我的问题是：在一个senarios中，消费者1得到一个消息，需要20秒来处理。在过程中间，他得到另一条消息，这条消息需要1秒来处理。他完成第二个消息处理，提交偏移量，然后立即崩溃。从而导致前面的消息处理失败。如果我正在运行消费者，他不会再读取第一条消息，因为第二条消息已经提交了大于第一条消息的offsst。我怎样才能避免这种情况呢？

Kafkaconsumer.on('message', async(message)=>{
await SOMETHING_ASYNC_1~20SEC;
Kafkaconsumer.commit(()=>{});
});

Answer 1

您基本上希望通过使用async.queue来限制消息和处理并发。

1. 创建具有消息处理器和并发的async.queue（消息处理器本身用setimmediate包装，因此它不会冻结事件循环）
2. 设置queue.drain以恢复使用者
3. 使用者的消息事件的处理程序暂停使用者并将消息推送到队列。

kafka-node自述文件在这里详细介绍了这一点。

在这里可以找到一个与您的问题类似的示例实现。