如何在kafka中创建新的消费者组

我按照快速入门指南上的说明在本地运行 kafkahere,

然后我在中定义了我的消费者组配置config/consumer.properties这样我的消费者就可以从定义的中选择消息group.id

运行以下命令，

bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

结果是，

test-consumer-group  <-- group.id defined in conf/consumer.properties
console-consumer-67807 <-- when connecting to kafka via kafka-console-consumer.sh

我能够通过基于 python 的消费者连接到 kafka，该消费者配置为使用提供group.id i.e test-consumer-group

首先，我无法理解kafka如何/何时创建消费者组。看来它加载了conf/consumer.properties在某个时间点，此外它还隐式地创建了消费者组（在我的例子中console-consumer-67807）通过连接时kafka-console-consumer.sh.

比方说，我如何明确创建自己的消费者组my-created-consumer-group ?

您不会显式创建消费者组，而是构建始终属于某个消费者组的消费者。无论您使用哪种技术（Spark、Spring、Flink...），每个 Kafka Consumer 都会有一个 Consumer Group。消费者组可以为每个单独的消费者进行配置。

似乎它在某个时间点加载conf/consumer.properties，此外，它在通过 kafka-console-consumer.sh 连接时隐式创建消费者组（在我的例子中是 console-consumer-67807）

如果您没有告诉您的控制台用户实际使用该文件，则不会考虑该文件。

有以下几种方法可以提供消费者组的名称：

带有属性文件的控制台 Consumer (--consumer.config)

文件是这样的config/consumer.properties应该看起来像

# consumer group id
group.id=my-created-consumer-group

这就是你如何确保控制台消费者接受这个group.id考虑到：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning --consumer.config /path/to/config/consumer.properties

使用 --group 控制台消费者

对于控制台消费者，消费者组会自动创建，带有前缀“console-consumer”和后缀（例如 PID），除非您通过添加来提供自己的消费者组--group:

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning --group my-created-consumer-group

基于标准代码的消费者 API

当使用标准 JAVA/Scala/... Consumer API 时，您可以通过以下属性提供 Consumer Group：

Properties settings = new Properties();
settings.put(ConsumerConfig.GROUP_ID_CONFIG, "basic-consumer");
// set more properties

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(settings)) {
consumer.subscribe(Arrays.asList("test-topic")