如何使用 kafka 生产者向同一主题发送消息?
How to send messages to same topic using kafka producer ?
我已经设置了生成和使用消息的 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();
可能是最琐碎的建议,但你确定你的第二条消息正在发送到 kafka 队列吗?
除了您的消费者,您可能希望在生产者发布消息时使用内置命令行消费者,以确保您的所有消息都已发布。
bin/kafka-console-consumer.sh --zookeeper host:2181 --topic test --from-beginning
我已经设置了生成和使用消息的 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();
可能是最琐碎的建议,但你确定你的第二条消息正在发送到 kafka 队列吗? 除了您的消费者,您可能希望在生产者发布消息时使用内置命令行消费者,以确保您的所有消息都已发布。
bin/kafka-console-consumer.sh --zookeeper host:2181 --topic test --from-beginning