Springboot整合kafka

Springboot 整合kafka

首先搭建两个springboot项目，一个作为生产者，一个作为消费者。kafka可以本地搭建，可以参考https://blog.csdn.net/weixin_42360600/article/details/123048409

生产者

项目结构

1、pom引入kafka依赖

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

2、application.properties中配置kafka生产者

server.port=8080
###########【Kafka集群】###########
spring.kafka.bootstrap-servers=localhost:9092
#==================================【初始化生产者配置】==================================#
# 重试次数
spring.kafka.producer.retries=0
# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
spring.kafka.producer.acks=1
# 批量大小
spring.kafka.producer.batch-size=16384
# 提交延时
spring.kafka.producer.properties.linger.ms=0
# 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
# linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了
# 生产端缓冲区大小
spring.kafka.producer.buffer-memory =33554432
# Kafka提供的序列化和反序列化类
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# 自定义分区器
#spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner

3、编写生产者发送消息

@ComponentpublicclassKafkaProducer{@AutowiredprivateKafkaTemplate<String,String> kafkaTemplate;/**
     * 发送消息
     */publicvoidsendMessage(){try{//生产消息String message ="hello ！ 测试kafka ";ListenableFuture<SendResult<String,String>> listenableFuture = kafkaTemplate.send("hello","hello", message);
            listenableFuture.addCallback(newListenableFutureCallback<SendResult<String,String>>(){@OverridepublicvoidonSuccess(SendResult<String,String> result){System.out.println("sendMessage success");}@OverridepublicvoidonFailure(Throwable ex){System.out.println("sendMessage error");}});}catch(Exception e){System.out.println("sendMessage exception");}}}

4、编写controller
页面请求触发kafka生产者发送消息

@RestControllerpublicclassKafkaProController{@AutowiredprivateKafkaProducer kafkaProducer;@RequestMapping("/hello")publicStringhello(){System.out.println("------->测试生产者发送消息");
        kafkaProducer.sendMessage();return"kafka消息已发送.";}}

消费者

项目结构

1、pom引入kafka

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

2、application.properties配置kafka消费者

server.port=8081
###########【Kafka集群】###########
spring.kafka.bootstrap-servers=localhost:9092
#==================================【初始化消费者配置】==================================#
# 默认的消费组ID
spring.kafka.consumer.properties.group.id=defaultConsumerGroup
# 是否自动提交offset
spring.kafka.consumer.enable-auto-commit=true
# 提交offset延时(接收到消息后多久提交offset)
spring.kafka.consumer.auto.commit.interval.ms=1000
# 当kafka中没有初始offset或offset超出范围时将自动重置offset
# earliest:重置为分区中最小的offset;
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
# none:只要有一个分区不存在已提交的offset,就抛出异常;
spring.kafka.consumer.auto-offset-reset=latest
# 消费会话超时时间(超过这个时间consumer没有发送心跳,就会触发rebalance操作)
spring.kafka.consumer.properties.session.timeout.ms=120000
# 消费请求超时时间
spring.kafka.consumer.properties.request.timeout.ms=180000
# Kafka提供的序列化和反序列化类
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# 消费端监听的topic不存在时，项目启动会报错(关掉)
spring.kafka.listener.missing-topics-fatal=false
# 设置批量消费
# spring.kafka.listener.type=batch
# 批量消费每次最多消费多少条消息
# spring.kafka.consumer.max-poll-records=50

3、编写消费者

@ComponentpublicclassTopicComponent{@KafkaListener(topics ={"hello"})publicvoidhandMessage(ConsumerRecord<String,String>record){String topic =record.topic();String msg =record.value();System.out.println("消费者接受消息：topic-->"+topic+",msg->>"+msg);}}

测试

先保证本地zookeeper 、kafka已经启动
启动生产者

启动消费者

页面访问，生产者发送消息

查看后台日志
生产者

消费者

ok，测试成功！结合具体业务编写msg即可