SpringBoot如何正确配置并运行Kafka
作者:学弟不想努力了
这篇文章主要介绍了SpringBoot如何正确配置并运行Kafka问题,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
一、配置pom.xml,引入maven依赖
<!-- 引入kafka依赖 --> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>2.8.6</version> </dependency>
二、application.yml配置文件
这里只提供了kafka有用的相关配置,其他的配置删了
spring: kafka: bootstrap-servers: xx.xx.xx.xx:9092 # kafka集群信息,多个用逗号间隔 # 生产者 producer: # 重试次数,设置大于0的值,则客户端会将发送失败的记录重新发送 retries: 3 batch-size: 16384 #批量处理大小,16K buffer-memory: 33554432 #缓冲存储大,32M acks: 1 # 指定消息key和消息体的编解码方式 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer # 消费者 consumer: # 消费者组 group-id: TestGroup # 是否自动提交 enable-auto-commit: false # 消费偏移配置 # none:如果没有为消费者找到先前的offset的值,即没有自动维护偏移量,也没有手动维护偏移量,则抛出异常 # earliest:在各分区下有提交的offset时:从offset处开始消费;在各分区下无提交的offset时:从头开始消费 # latest:在各分区下有提交的offset时:从offset处开始消费;在各分区下无提交的offset时:从最新的数据开始消费 auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer # 监听 listener: # record:当每一条记录被消费者监听器(ListenerConsumer)处理之后提交 # batch:当每一批poll()的数据被ListenerConsumer处理之后提交 # time:当每一批poll()的数据被ListenerConsumer处理之后,距离上次提交时间大于TIME时提交 # count:当每一批poll()的数据被ListenerConsumer处理之后,被处理record数量大于等于COUNT时提交 # count_time:TIME或COUNT中有一个条件满足时提交 # manual:当每一批poll()的数据被ListenerConsumer处理之后, 手动调用Acknowledgment.acknowledge()后提交 # manual_immediate:手动调用Acknowledgment.acknowledge()后立即提交,一般推荐使用这种 ack-mode: manual_immediate
三、消费者
消费者监听,可以配置多个监听器
import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.stereotype.Component; /** * 消费者 * kafka监听器 */ @Component public class KafkaConsumer { /** * kafka的监听器1,topic为"topic_test",消费者组为"group_topic_test" * @param record * @param item */ @KafkaListener(topics = "topic_test", groupId = "group_topic_test") public void topicListener1(ConsumerRecord<String, String> record, Acknowledgment item) { String value = record.value(); System.out.println(value); System.out.println(record); //手动提交 item.acknowledge(); } /** * 配置多个消费组 * kafka的监听器2,topic为"topic_test2",消费者组为"group_topic_test" * @param record * @param item */ @KafkaListener(topics = "topic_test2",groupId = "group_topic_test2") public void topicListener2(ConsumerRecord<String, String> record, Acknowledgment item) { String value = record.value(); System.out.println(value); System.out.println(record); item.acknowledge(); } }
四、生产者
生产者作为接口Api作为测试
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; /** * kafka生产者 */ @RestController @RequestMapping("/kafka") public class KafkaController { @Autowired private KafkaTemplate<String, String> kafkaTemplate; @RequestMapping("/send") public void send() { kafkaTemplate.send("topic_test", "key", "测试kafka消息"); } }
五、调用测试
启动Boot项目,使用Postman工具发送GET请求:
http://localhost:8080/kafka/send
总结
以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。