我在使用消费群消费kafka消息时收到重复消息。
我正在使用这个nodejs库。https://www.npmjs.com/package/kafka-node
我的消费代码如下
const config = require( '../../configs' );
const kafka = require( 'kafka-node' );
var options = {
id: 'consumer1',
kafkaHost: config.kafka.prod.kafka_host, //multiple kafka hosts (comma separated)
groupId: "test-group2",
sessionTimeout: 15000,
protocol: ['roundrobin'],
fromOffset: 'earliest'
};
var consumerGroup = new kafka.ConsumerGroup( options, 'my-replicated-topic3' );
consumerGroup.on( 'message', function ( message ) {
console.log( message );
} );
我的成绩差了。
{ topic: 'my-replicated-topic3',
value: '{"meta":{"topic":"my-replicated-topic3","added_at":"2019-02-22T09:25:54.708Z","server":"cron"},"data":"1550827554708 ::: Totam quis qui. Sit dolore laboriosam odio. Facilis porro et quam repellat pariatur. Ad voluptatem quidem."}',
offset: 8941,
partition: 0,
highWaterOffset: 8966,
key: null }
---
---
---
---
{ topic: 'my-replicated-topic3',
value: '{"meta":{"topic":"my-replicated-topic3","added_at":"2019-02-22T09:25:54.708Z","server":"cron"},"data":"1550827554708 ::: Totam quis qui. Sit dolore laboriosam odio. Facilis porro et quam repellat pariatur. Ad voluptatem quidem."}',
offset: 8941,
partition: 0,
highWaterOffset: 8970,
key: null }
您可以看到,每隔几次记录之后,相同的消息就会重复出现。在这里,消息的偏移量是相同的,但highwateroffset与所有重复消息的偏移量不同。
请提出解决方法。
1条答案
按热度按时间js81xvg61#
您是否重新示例化您的消费者?您的消费者如何承诺其补偿?
默认情况下,消费者每5秒自动提交其偏移量。你的图书馆就是这样。
如果在提交偏移量之前重新示例化使用者,它将从上次提交的偏移量重新启动。