使用Spring Boot的多线程事务Kafka制作者和消费者
我有一个使用Spring Boot相关项目的项目。我想在项目中使用Kafka消费者和生产者的Transactional功能。我需要尽可能高效地在Kafka中生成大量消息。所以我需要一个多线程消费和生产来满足这个要求。如何使用Spring boot开发多线程消费者和生产者？
共有1个答案
请参阅并发侦听器启动属性。
spring.kafka.listener.concurrency
主题必须至少具有与并发性相同的分区。
https://docs.spring.io/spring-kafka/docs/2.6.1/reference/html/#message-侦听器容器
-
Kafka的doc给出了一种方法，大约用以下描述： 每个线程一个消费者：一个简单的选择是为每个线程提供自己的消费者 我的代码： 但它不起作用，引发了一个异常： JAVAutil。ConcurrentModificationException:KafkaConsumer对于多线程访问不安全 此外，我还阅读了Flink（一个用于分布式流和批处理数据的开源平台）的源代码。Flink使用多线程消费程序与我
-
我正在使用Spring Kafka消费者。我已将并发设置为10，并创建了5个消费者（用于5个主题）。所以有50个Spring Kafka消费者线程。 Kafka消费者可以使用的最大线程数是多少？如何增加此线程池的大小？我查阅了spring文档，但没有发现任何相关内容。
-
我有一个Spring-boot应用程序，可以听Kafka。为了避免重复处理，我尝试手动提交。为此，我在阅读主题后异步提交了一条消息。但是我被困在如何实现消费者幂等，这样记录就不会被处理两次。
-
我编写了一个Kafka消费者（使用Spring Kafka），它从单个主题读取并且是消费者组的一部分。一旦消息被消费，它将执行所有下游操作并继续下一个消息偏移量。我已将其打包为WAR文件，我的部署管道将其推送到单个实例。使用我的部署管道，我可能可以将此工件部署到我的部署池中的多个实例。 但是，当我希望多个消费者作为我的基础设施的一部分时，我无法理解以下内容- > 实际上，我可以在部署池中定义多个实
-
Flink kafka消费者有两种类型的消费者，例如： 这两个消费者层次结构扩展了相同的类。我想知道维护编号类背后的设计决策是什么？我们什么时候应该使用其中一种？ 我注意到带有数字后缀的类有更多的特性（例如ratelimiting）。 https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka
-
我使用的是高级消费者，如所述：https://cwiki.apache.org/confluence/display/KAFKA/Consumer组示例 我注意到，我的消费者不会永远运行，而是在一段时间后结束。在zookeeper一侧，我看到以下内容： INFO已处理的会话终止的setsionid： 0x144a4854325004d（org.apache.zookeeper.server.准备请