尝试在 Kafka 中设置消费者和生产者

Question

我正在尝试使用 node-rdkafka 在 Kafka 上创建一个简单的生产者-消费者流

我正在使用debug: 'all' 模式，这是我从日志中得到的：

制作人：test [0]: MessageSet with 1 message(s) delivered

消费者：Fetch topic test [0] at offset 38 (v2)

消费者在生成消息时正在更改偏移量这一事实使我相信与代理的连接已正确设置和验证。

但是，由于某种原因，我没有在消费者中收到消息本身。

永远不会调用此事件：

consumer.on('data', function(m) {
    console.log("consumed", m)
});

我创建了一个demo项目进行测试，你需要有一个支持SASL_SSL协议的Kafka broker才能使用：

https://github.com/guysegal/kafka-example

具体来说，这是消费者代码：

https://github.com/guysegal/kafka-example/blob/master/src/consumer.ts

和生产者代码：

https://github.com/guysegal/kafka-example/blob/master/src/producer.ts

Answer 1

您可能希望设置值'auto.offset.reset': 'earliest' 并再次启动消费者。

auto.offset.reset 

这个属性决定了消费者的起始位置，以防消费者从没有最后一个已知状态的组开始。例如如果是新的 group-id id。

更改auto.offset.reset 的值的原因是因为如果您先生产然后再启动消费者，那么主题偏移量已经增加，消费者将从最新的偏移量（增加的偏移量）开始并从该点开始读取消息。

而在'earliest' 的情况下，消费者从关于主题的第一条可用消息开始。