Springboot整合kafka_音乐土豆_springboot整合kafka

未知 02-07 8076

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、编写生产者发送消息

@Component public class KafkaProducer { @Autowired private KafkaTemplate<String,String> kafkaTemplate; /** * 发送消息 */ public void sendMessage() { try{ //生产消息 String message = "hello ！测试kafka "; ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send("hello","hello", message); listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() { @Override public void onSuccess(SendResult<String, String> result) { System.out.println("sendMessage success"); } @Override public void onFailure(Throwable ex) { System.out.println("sendMessage error"); } }); }catch (Exception e){ System.out.println("sendMessage exception"); } } }

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

@RestController public class KafkaProController { @Autowired private KafkaProducer kafkaProducer; @RequestMapping("/hello") public String hello(){ System.out.println("------->测试生产者发送消息"); kafkaProducer.sendMessage(); return "kafka消息已发送."; } } 消费者

项目结构 1、pom引入kafka

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

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、编写消费者

@Component public class TopicComponent { @KafkaListener(topics = {"hello"}) public void handMessage(ConsumerRecord<String, String> record){ String topic = record.topic(); String msg = record.value(); System.out.println("消费者接受消息：topic-->"+topic+",msg->>"+msg); } } 测试

先保证本地zookeeper 、kafka已经启动启动生产者启动消费者页面访问，生产者发送消息查看后台日志生产者消费者 ok，测试成功！结合具体业务编写msg即可