springboot 集成 kafka批量消费数据-Toy模板网

这篇具有很好参考价值的文章主要介绍了springboot 集成 kafka批量消费数据。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

新建一个maven工程，添加kafka依赖

	<dependency>
		<groupId>org.springframework.kafka</groupId>
		<artifactId>spring-kafka</artifactId>
	</dependency>

yaml配置文件文章来源地址https://www.toymoban.com/news/detail-539399.html

spring:
	kafka:
	    bootstrap-servers: ${local_host_ip}:9092
	    producer: # producer 生产者
	      retries: 0 # 重试次数
	      acks: 1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
	      batch-size: 100000 # 批量处理的最大大小 单位 byte
	      buffer-memory: 33554432 # 生产端缓冲区大小
	      key-serializer: org.apache.kafka.common.serialization.StringSerializer
	      value-serializer: org.apache.kafka.common.serialization.StringSerializer
	    consumer: # consumer消费者
	      group-id: javagroup # 默认的消费组ID
	      enable-auto-commit: false # 是否自动提交offset
	      #auto-commit-interval: 100  # 提交offset延时(接收到消息后多久提交offset)
	      # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
	      # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
	      # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
	      auto-offset-reset: latest
	      max-poll-records: 1000   #批量消费每次最多消费多少条消息
	      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
	      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
	    listener:
	      enable-auto-commit: false
	      ack-mode: manual_immediate
	      max-poll-records: 1000
	      # 消费者监听的topic不存在时，项目会报错，设置为false
	      missing-topics-fatal: false
	      type: batch

	/**
	 * 手动提交批量消费
	 *
	 * @param list (list 的大小和 batch-size（单位） 有关)
	 * @param consumer
	 */
	@KafkaListener(topics = "topic_test", groupId = "topic_test_group")
	public void recieveRealTimNta(List<Message<String>> list, Consumer consumer) {
		try {
			//业务模块
			consumer.commitAsync();
		} catch (Exception e) {
			log.error("安全事件-告警数据解析 -nta  消费失败");
		} finally {
			consumer.commitSync();
		}
	}