如何使用Kafka制作人将消息发送到同一主题？

Question

我已经设置了kafka客户端，它可以产生和消费消息，当我们把有效载荷从生产者发送到主题时，它可以正常工作，所以我有问题生产者现在第一个消息我可以发送到主题，我也可以从kafka主题中消费，现在我尝试发送第二个消息，但是消费者没有从kafka主题中读取第二个消息，知道这里发生了什么吗？

Producer.js

var config = require('./config.js');
var zk = require('node-zookeeper-client');
var kafkaConn = "130.8";
var kafka = require('kafka-node'),
    HighLevelProducer = kafka.HighLevelProducer,
    client = new kafka.Client(kafkaConn),
    producer = new HighLevelProducer(client),
    payloads = [
        { topic: 'test', messages: 'second message' }
    ];
producer.on('ready', function () {
    producer.send(payloads, function (err, data) {
        console.log(data);
    });
});

consumer.js

function start () {
    topics = [{topic: 'test'}];
    var groupId = 'ulogGroup';
    var clientId = "consumer-" + Math.floor(Math.random() * 10000);
    var options = {autoCommit: true, fetchMaxWaitMs: 100, fetchMaxBytes: 10 * 1024 * 1024, groupId: groupId};
    console.log("Started consumer: ", clientId);
    var consumer_client = new kafka.Client(kafkaConn,clientId);
    var client = new Client(consumer_client.connectionString,clientId);
    var consumer = new HighLevelConsumer(client, topics, options);
    console.log("Consumer topics:", getConsumerTopics(consumer).toString());
   // startConsumer(consumer);
    consumer.on('message', function (message) {
        //var topic = message.data;
        console.log('Message',message);
    });
};
start();

Answer 1

可能是最微不足道的建议，但你确定你的第二条消息被发送到Kafaka队列吗？除了您的使用者之外，您可能希望在生产者发布消息时使用内置的命令行使用者，以确保所有消息都已发布。

bin/kafka-console-consumer.sh --zookeeper host:2181 --topic test --from-beginning