【问题标题】:Not able to read event from single topic that has single partition through multiple consumers无法通过多个消费者从具有单个分区的单个主题读取事件
【发布时间】:2018-11-06 10:14:44
【问题描述】:

我是 Kafka 的新手,使用 Kafka-node。我正在使用 Kafka 进行实时数据处理。 该系统有一个生产者和多个消费者。如果有任何一个消费者同时运行,我能够接收数据,但如果我运行两个消费者,则只有一个消费者会接收数据,而另一个消费者不会收到数据。

kafka 生产者的代码是:

const config = require('../config');
const logger = require('./logger');
const kafka = require('kafka-node'),
HighLevelProducer = kafka.HighLevelProducer,
client = new kafka.Client(`${config.kafka.host}:${config.kafka.port}`),
producer = new HighLevelProducer(client);

producer.on('ready', () => {
logger.info("Events producer to kafka is ready...");
});

producer.on('error', (err) => {
logger.error("Error while starting kafka producer:" + err.message);
});

const queueEvent = (event, callback) => {
const payloads = [
{ topic: config.kafka.queueName, messages: JSON.stringify(event, null, 
2) },
];
producer.send(payloads, (err, data) => {
if (err) {
logger.error(`Error while producing data: ${err.message}`);
callback(err);
} else {
callback(null, data);
}
});
};

module.exports = {
queueEvent
};

为所有消费者完成的配置如下所示:

const kafka = require('kafka-node');

const logger = require('../common/logger');
const config = require('../common/config');
const eventDao = require('../models/event');
const _ = require('lodash');

const { getDeviceHierarchy } = require('../common/apiUtils');

const options = { autoCommit: true, fetchMaxWaitMs: 1000, fetchMaxBytes: 
1024 * 1024 };

const HighLevelConsumer = kafka.HighLevelConsumer,
client = new kafka.Client(`${config.kafka.host}:${config.kafka.port}`),
consumer = new HighLevelConsumer(
client,
[
{ topic: config.kafka.queueName, partition: 0 }
],
options
);

我使用的是 Kafka 的 docker 镜像,下面是我所做的设置

docker run -d -p 2181:2181 -p 3030:3030 -p 8081-8083:8081-8083 -p 9581-9585:9581-9585 -p 9092:9092 -e ADV_HOST=localhost -e DISABLE=azure -documentdb,blockchain,bloomberg,cassandra,coap,druid,elastic,elastic5,ftp,hazelcast,hbase,influxdb,jms,kudu,mongodb,mqtt,redis,rethink,voltdb,yahoo,hdfs,jdbc,elasticsearch,s3,twitter -e CONNECT_HEAP=6G -e RUNNING_SAMPLEDATA=0 -e RUNTESTS=0 landoop/fast-data-dev:latest**

【问题讨论】:

    标签: apache-kafka


    【解决方案1】:

    请确认多消费场景中的消费者是否属于同一个消费组?

    如果他们这样做,那么观察到的行为是正确的。让我试着详细说明一下:

    1. 在您描述的场景中,听起来两个消费者都属于同一个消费者组。在这种情况下,组中的每个消费者只能从一个分区消费。因为我们这里只有一个分区,所以排队的第一个消费者会消费它。

    2. 如果我们有多个消费者组,每个组中有多个消费者订阅同一个主题(有一个分区)。在这种情况下,多个消费者可以从同一个分区消费。

    我不熟悉您使用的编程语言,但我在您的 Kafka 使用者中找不到设置“group.id”属性的语句。您能否尝试在您的消费者代码/配置中进行设置?

    另外,您能否检查并确认您正在使用的 Kafka 版本以及 Kafka 的 consumer.properties 文件中是否有默认值?因为从版本 0.9.0.0 开始,group.id 属性值已被强制设置,不提供此值会引发错误(检查此ticket)。

    【讨论】:

    • 谢谢。消费者属于同一个消费者群体。创建不同的消费者组已经解决了这个问题。
    • 但现在我面临另一个问题:
    • @monika - 好的。好吧,如果这个问题解决了。你能承认/接受这个答案吗? :) (您也将从确认中受益:))然后针对其他问题发布一个新问题并提供相同的链接?然后,我将尝试重现这些新问题并进行检查。
    猜你喜欢
    • 1970-01-01
    • 2021-11-28
    • 2020-05-03
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多