即时通讯使用模块kafka-node https://github.com/SOHU-Co/kafka-node

每当我重新启动消费者时,他们都会收到所有旧消息,即使用循环系统(负载均衡)

你有什么想法我怎么能向服务器宣布我消费了一条消息,当我重新启动消费者时他不再发送给我了？

我的代码或配置服务器有些错误？

任何的想法 ？

生产者代码

var kafka = require('kafka-node');
var HighLevelProducer = kafka.HighLevelProducer;
var Client = kafka.Client;
var client = new Client('xxx.xxx.xxx.xxx:2181,xxx.xxx.xxx.xxx:2181,xxx.xxx.xxx.xxx:2181,xxx.xxx.xxx.xxx:2181,xxx.xxx.xxx.xxx:2181', 'consumer' + process.pid);
var argv = require('optimist').argv;
var topic = argv.topic || 'test_12345';
var producer = new HighLevelProducer(client);
var time = process.hrtime();

var message, diff,i=0;
producer.on('ready', function () {
    setInterval(function(){
        var date = new Date();
        var dateString = date.getFullYear() + "-" +((date.getMonth()+1)<10 ? '0'+(date.getMonth()+1) : (date.getMonth()+1)) + "-" +(date.getDate()<10 ? '0'+date.getDate() : date.getDate()) + " …