Spring Boot中怎么使用@KafkaListener并发批量接收消息

云天博客 • 05/14 20:51 • 开发 • 阅读 25

这篇“SpringBoot中怎么使用@KafkaListener并发批量接收消息”文章的知识点大部分人都不太理解，所以小编给大家总结了以下内容，内容详细，步骤清晰，具有一定的借鉴价值，希望大家阅读完这篇文章能有所收获，下面我们一起来看看这篇“SpringBoot中怎么使用@KafkaListener并发批量接收消息”文章吧。###第一步，并发消费###
先看代码，重点是这我们使用的是ConcurrentKafkaListenerContainerFactory并且设置了factory.setConcurrency(4); （我的topic有4个分区，为了加快消费将并发设置为4，也就是有4个KafkaMessageListenerContainer）注意也可以直接在application.properties中添加spring.kafka.listener.concurrency=3，然后使用@KafkaListener并发消费。###第二步，批量消费###
然后是批量消费。重点是factory.setBatchListener(true);
以及 propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
一个设启用批量消费，一个设置批量消费每次最多消费多少条消息记录。重点说明一下，我们设置的ConsumerConfig.MAX_POLL_RECORDS_CONFIG是50，并不是说如果没有达到50条消息，我们就一直等待。官方的解释是”The maximum number of records returned in a single call to poll().”，也就是50表示的是一次poll最多返回的记录数。从启动日志中可以看到还有个 max.poll.interval.ms = 300000，也就说每间隔max.poll.interval.ms我们就调用一次poll。每次poll最多返回50条记录。max.poll.interval.ms官方免费云主机域名解释是”The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member. “;启动日志截图关于max.poll.records和max.poll.interval.ms官方解释截图：###第三步，分区消费###
对于只有一个分区的topic，不需要分区消费，因为没有意义。下面的例子是针对有2个分区的情况（我的完整代码中有4个listenPartitionX方法，我的topic设置了4个分区），读者可以根据自己的情况进行调整。以上就是关于“SpringBoot中怎么使用@KafkaListener并发批量接收消息”这篇文章的内容，相信大家都有了一定的了解，希望小编分享的内容对大家有帮助，若想了解更多相关的知识内容，请关注百云主机行业资讯频道。

相关推荐: Python中Selenium上传文件的方式有哪些

今天小编给免费云主机域名大家分享一下Python中Selenium上传文件的方式有哪些的相关知识点，内容详细，逻辑清晰，相信大部分人都还太了解这方面的知识，所以分享这篇文章给大家参考一下，希望大家阅读完这篇文章后有所收获，下面我们一起来了解一下吧。如果页面需要…