entityMap|blocks|key|1vm4g|text|您必须使用ConcurrentMessageListenerContainer。它委托一个或多个KafkaMessageListenerContainer实例来提供多线程消费。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|3pdi0|@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,+String>>+kafkaListenerContainerFactory()+{
++++ConcurrentKafkaListenerContainerFactory<String,+String>+factory+=
++++++++new+ConcurrentKafkaListenerContainerFactory<>();
++++factory.setConsumerFactory(consumerFactory());
++++factory.setConcurrency(10);
++++factory.getContainerProperties().setPollTimeout(3000);
++++return+factory;
}|code-block|syntax|javascript|d7pbb|factory.setConcurrency(+10+)创建10个KafkaMessageListenerContainer实例。每个实例都会获得一定数量的分区。这取决于您在创建topic时配置的分区数量。|21oq7|一些准备步骤：|106th|@Autowired
private+KafkaTemplate<String,+Object>+kafkaTemplate;

private+final+static+String+BOOTSTRAP_ADDRESS+=+"localhost:9092";
private+final+static+String+CONSUMER_GROUP+=+"consumer-group-1";
private+final+static+String+TOPIC+=+"test-topic";

@Bean
public+ConsumerFactory<String,+String>+consumerFactory()+{
++++Map<String,+Object>+props+=+new+HashMap<>();
++++props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,+BOOTSTRAP_ADDRESS);
++++props.put(ConsumerConfig.GROUP_ID_CONFIG,+CONSUMER_GROUP);
++++props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,+StringDeserializer.class);
++++props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,+StringDeserializer.class);
++++return+new+DefaultKafkaConsumerFactory<>(props);
}


@KafkaListener(topics+=+TOPIC,+containerFactory+=+"kafkaListenerContainerFactory")
public+void+listen(@Payload+String+message)+{
++++logger.info(message);
}

public+void+start()+{
++++try+{
++++++++Thread.sleep(5000L);
++++}+catch+(InterruptedException+e)+{
++++++++e.printStackTrace();
++++}
++++
++++for+(int+i+=+0;+i+<+10;+i%2B%2B)+{
++++++++kafkaTemplate.send(TOPIC,+i,+String.valueOf(i),+"Message+"+%2B+i);
++++}
++++
++++logger.info("All+message+are+sent");
}|fnkpf|如果运行上面的方法，您可以看到每个KafkaMessageListenerContainer实例都在处理放入该实例所服务的分区中的消息。添加Thread.sleep()以等待消费者被初始化。|4gog9|2020-07-01+15:48:34.800++INFO+201566+---+[ntainer#0-4-C-1]+r.s.c.KafkaConsumersDemo+++++++++++++++++:+Message+5
2020-07-01+15:48:34.801++INFO+201566+---+[ntainer#0-6-C-1]+r.s.c.KafkaConsumersDemo+++++++++++++++++:+Message+7
2020-07-01+15:48:34.800++INFO+201566+---+[ntainer#0-7-C-1]+r.s.c.KafkaConsumersDemo+++++++++++++++++:+Message+8
2020-07-01+15:48:34.800++INFO+201566+---+[ntainer#0-9-C-1]+r.s.c.KafkaConsumersDemo+++++++++++++++++:+Message+1
2020-07-01+15:48:34.800++INFO+201566+---+[ntainer#0-0-C-1]+r.s.c.KafkaConsumersDemo+++++++++++++++++:+Message+0
2020-07-01+15:48:34.800++INFO+201566+---+[ntainer#0-8-C-1]+r.s.c.KafkaConsumersDemo+++++++++++++++++:+Message+9
2020-07-01+15:48:34.800++INFO+201566+---+[ntainer#0-3-C-1]+r.s.c.KafkaConsumersDemo+++++++++++++++++:+Message+4
2020-07-01+15:48:34.801++INFO+201566+---+[ntainer#0-2-C-1]+r.s.c.KafkaConsumersDemo+++++++++++++++++:+Message+3
2020-07-01+15:48:34.801++INFO+201566+---+[ntainer#0-1-C-1]+r.s.c.KafkaConsumersDemo+++++++++++++++++:+Message+2
2020-07-01+15:48:34.800++INFO+201566+---+[ntainer#0-5-C-1]+r.s.c.KafkaConsumersDemo+++++++++++++++++:+Message+6|67n8r^0|5|Y|1C|T|0|0|X|T|0|0|0|H|T|0|0^^$0|$]|1|@$2|3|4|5|6|7|8|W|9|@$A|X|B|Y|C|D]|$A|Z|B|10|C|D]]|E|@]|F|$]]|$2|G|4|H|6|I|8|11|9|@]|E|@]|F|$J|K]]|$2|L|4|M|6|7|8|12|9|@$A|13|B|14|C|D]]|E|@]|F|$]]|$2|N|4|O|6|7|8|15|9|@]|E|@]|F|$]]|$2|P|4|Q|6|I|8|16|9|@]|E|@]|F|$J|K]]|$2|R|4|S|6|7|8|17|9|@$A|18|B|19|C|D]]|E|@]|F|$]]|$2|T|4|U|6|I|8|1A|9|@]|E|@]|F|$J|K]]|$2|V|4|-4|6|7|8|1B|9|@]|E|@]|F|$]]]]

You have to use <code>ConcurrentMessageListenerContainer</code>. It delegates to one or more <code>KafkaMessageListenerContainer</code> instances to provide multi-threaded consumption.
<pre class="lang-java prettyprint-override"><code>@Bean
KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;String, String&gt;&gt; kafkaListenerContainerFactory() {
 ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory =
 new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
 factory.setConsumerFactory(consumerFactory());
 factory.setConcurrency(10);
 factory.getContainerProperties().setPollTimeout(3000);
 return factory;
}
</code></pre>
factory.setConcurrency(10) creates 10 <code>KafkaMessageListenerContainer</code> instances. Each instance gets some amount of partitions. It depends on the number of partitions you configured when you created the topic.
Some preparation steps:
<pre class="lang-java prettyprint-override"><code>@Autowired
private KafkaTemplate&lt;String, Object&gt; kafkaTemplate;

private final static String BOOTSTRAP_ADDRESS = &quot;localhost:9092&quot;;
private final static String CONSUMER_GROUP = &quot;consumer-group-1&quot;;
private final static String TOPIC = &quot;test-topic&quot;;

@Bean
public ConsumerFactory&lt;String, String&gt; consumerFactory() {
 Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
 props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_ADDRESS);
 props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP);
 props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
 props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
 return new DefaultKafkaConsumerFactory&lt;&gt;(props);
}


@KafkaListener(topics = TOPIC, containerFactory = &quot;kafkaListenerContainerFactory&quot;)
public void listen(@Payload String message) {
 logger.info(message);
}

public void start() {
 try {
 Thread.sleep(5000L);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 
 for (int i = 0; i &lt; 10; i++) {
 kafkaTemplate.send(TOPIC, i, String.valueOf(i), &quot;Message &quot; + i);
 }
 
 logger.info(&quot;All message are sent&quot;);
}
</code></pre>
If you run the method above you can see that each <code>KafkaMessageListenerContainer</code> instance processes the messages being put into the partition which that instance serves.
Thread.sleep() is added to wait for the consumers to be initialized.
<pre class="lang-java prettyprint-override"><code>2020-07-01 15:48:34.800 INFO 201566 --- [ntainer#0-4-C-1] r.s.c.KafkaConsumersDemo : Message 5
2020-07-01 15:48:34.801 INFO 201566 --- [ntainer#0-6-C-1] r.s.c.KafkaConsumersDemo : Message 7
2020-07-01 15:48:34.800 INFO 201566 --- [ntainer#0-7-C-1] r.s.c.KafkaConsumersDemo : Message 8
2020-07-01 15:48:34.800 INFO 201566 --- [ntainer#0-9-C-1] r.s.c.KafkaConsumersDemo : Message 1
2020-07-01 15:48:34.800 INFO 201566 --- [ntainer#0-0-C-1] r.s.c.KafkaConsumersDemo : Message 0
2020-07-01 15:48:34.800 INFO 201566 --- [ntainer#0-8-C-1] r.s.c.KafkaConsumersDemo : Message 9
2020-07-01 15:48:34.800 INFO 201566 --- [ntainer#0-3-C-1] r.s.c.KafkaConsumersDemo : Message 4
2020-07-01 15:48:34.801 INFO 201566 --- [ntainer#0-2-C-1] r.s.c.KafkaConsumersDemo : Message 3
2020-07-01 15:48:34.801 INFO 201566 --- [ntainer#0-1-C-1] r.s.c.KafkaConsumersDemo : Message 2
2020-07-01 15:48:34.800 INFO 201566 --- [ntainer#0-5-C-1] r.s.c.KafkaConsumersDemo : Message 6
</code></pre>

entityMap|blocks|key|d6saq|text|是的，@KafkaListener将为您创建多个消费者。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|e9fgu|这样，您就可以将它们全部配置为使用相同的主题并属于相同的组。Kafka协调器将把分区分发给你的消费者。|9kmmh|不过，如果在本主题中只有一个分区，则不会发生并发:单个分区在单个线程中处理。|dh2gd|另一种选择实际上是配置一个concurrency，并且将根据concurrency+<->+partition状态再次创建几个使用者。|dfh6t^0|3|E|0|0|0|D|B|U|P|0^^$0|$]|1|@$2|3|4|5|6|7|8|N|9|@$A|O|B|P|C|D]]|E|@]|F|$]]|$2|G|4|H|6|7|8|Q|9|@]|E|@]|F|$]]|$2|I|4|J|6|7|8|R|9|@]|E|@]|F|$]]|$2|K|4|L|6|7|8|S|9|@$A|T|B|U|C|D]|$A|V|B|W|C|D]]|E|@]|F|$]]|$2|M|4|-4|6|7|8|X|9|@]|E|@]|F|$]]]]

Yes, the <code>@KafkaListener</code> will create multiple consumers for you.
With that you can configure all of them to use the same topic and belong to the same group.
The Kafka coordinator will distribute partitions to your consumers.
Although if you have only one partition in the topic, the concurrency won't happen: a single partition is processed in a single thread.
Another option is indeed to configure a <code>concurrency</code> and again several consumers are going to be created according <code>concurrency &lt;-&gt; partition</code> state.

entityMap|blocks|key|178hv|text|正如@Salavat+Yalalo建议的那样，我将我的卡夫卡容器工厂设为ConcurrentKafkaListenerContainerFactory。在@KafkaListenere方法中，我添加了一个名为concurrency的选项，它接受一个整数作为字符串，指示要跨越的消费者数量，如下所示|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|aetn2|@KafakListener(concurrency+="4",+containerFactory="concurrentKafkaListenerContainerFactory(bean+name+of+the+factory)",..other+optional+values)
public+void+topicConsumer(Message<MyObject>+myObject){
//.....
}|code-block|syntax|javascript|18b4l|当运行时，我看到在一个消费者组中创建了4个消费者。|5ii2u^0|10|13|0|0|0^^$0|$]|1|@$2|3|4|5|6|7|8|O|9|@$A|P|B|Q|C|D]]|E|@]|F|$]]|$2|G|4|H|6|I|8|R|9|@]|E|@]|F|$J|K]]|$2|L|4|M|6|7|8|S|9|@]|E|@]|F|$]]|$2|N|4|-4|6|7|8|T|9|@]|E|@]|F|$]]]]

As @Salavat Yalalo suggested I made my Kafka container factory to be <code>ConcurrentKafkaListenerContainerFactory</code>. On the @KafkaListenere method I added option called concurrency which accepts an integer as a string which indicates number of consumers to be spanned, like below
<pre><code>@KafakListener(concurrency =&quot;4&quot;, containerFactory=&quot;concurrentKafkaListenerContainerFactory(bean name of the factory)&quot;,..other optional values)
public void topicConsumer(Message&lt;MyObject&gt; myObject){
//.....
}
</code></pre>
When ran, I see 4 consumers being created in a single consumer group.

I am trying to create multiple consumers in a consumer group for parallel processing since we have heavy inflow of messages. I am using spring boot and KafkTemplate. How can we create multiple consumers belonging to single consumer group, in single instance of spring boot application?
Does having multiple methods annotated with @KafkaListener will create multiple consumers?

How to create multiple consumers in a consume group using Spring provided Kafka apis

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

 我试图在一个消费者组中创建多个消费者，以便进行并行处理，因为我们有大量的消息流入。我使用的是spring boot和KafkTemplate。如何在spring boot应用程序的单个实例中创建属于单个消费者组的多个消费者？使用@KafkaListener注释多个方法会创建多个消费者吗？ 

问如何使用Spring提供的Kafka apis在一个消费组中创建多个消费者
EN

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何使用Spring提供的Kafka apis在一个消费组中创建多个消费者EN