如何在Kafka中创建一个新的消费者组。

Question

如何在Kafka中创建一个新的消费者组。

10

我按照快速入门指南在此本地运行了kafka，

然后我在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，即test-consumer-group。

首先，我不理解 Kafka 是如何/何时创建消费者组的。它似乎在某个时刻加载conf/consumer.properties文件，并在通过kafka-console-consumer.sh进行连接时隐式地创建消费者组（在我的情况下是console-consumer-67807）。

我该如何显式地创建自己的消费者组，比如说my-created-consumer-group？

- laxman

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Michael Heil · Accepted Answer

您不需要显式地创建消费者组，而是构建属于消费者组的消费者。无论使用哪种技术（Spark、Spring、Flink等），每个Kafka消费者都将拥有一个消费者组。消费者组可针对每个单独的消费者进行配置。

它似乎在某个时间点加载了conf/consumer.properties文件，并在通过kafka-console-consumer.sh连接时隐式创建了消费者组（在我的情况下为console-consumer-67807）

如果您没有告诉控制台消费者实际利用该文件，它就不会被考虑在内。

以下是提供消费者组名称的几种替代方法：

带有属性文件的控制台消费者（--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的控制台消费者

对于控制台消费者，如果您不提供自己的消费者组（通过添加--group选项），则消费者组将自动以前缀为“console-consumer”和类似于PID的后缀创建：

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

标准基于代码的消费者API

使用标准的JAVA/Scala/...消费者API时，您可以通过属性提供消费者组：

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")