spring boot整合kafka记录

心已赠人 2021-09-24 11:12 368阅读 0赞

pom文件

  1. <dependencies>
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter-web</artifactId>
  5.         </dependency>
  6.  <dependency>
  7.             <groupId>org.springframework.kafka</groupId>
  8.             <artifactId>spring-kafka</artifactId>
  9.         </dependency>

yml配置

  1. spring：
  2.   kafka:
  3. #    bootstrap-servers: 192.168.10.45:9092
  4.     bootstrap-servers: 192.168.198.128:9092
  5.     producer:
  6.       # 重试次数，默认Integer.MAX_VALUE
  7.       retries: 1
  8.       # 同一批次内存大小（默认16K）
  9.       batch-size: 16384
  10.       # 生产者内存缓存区大小(32M)
  11.       buffer-memory: 33554432
  12.       # key和value的序列化（默认，可以不设置）
  13.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  14.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  15.       # ack应答机制，默认1，即只需要确认leader收到消息
  16.       acks: 1
  17.         # springboot1.5.16自动装配中不支持properties下的其他配置，不知道为啥。2.x版本可以
  18.         #properties:
  19.       # 使用自定义的分区选择器
  20.       #{partitioner.class: com.msy.kafka.MyPartition, acks: all}
  21.     consumer:
  22.       group-id: test
  23.       enable-auto-commit: false
  24.       # earliest：从头开始消费   latest：从最新的开始消费   默认latest
  25.       auto-offset-reset: latest
  26.       # key和value反序列化（默认，可以不设置）
  27.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  28.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  29.  #最大消费数
  30.       max-poll-records: 100  
  31.     listener:
  32.       # 消费者并发能力  这个并发量根据分区数决定，必须小于等于分区数，否则会有线程一直处于空闲状态 创建一个分区数为8的Topic。
  33.       concurrency: 6
  34.       # 设置手动提交的时候，需要设置ackMode
  35.       ack-mode: MANUAL
  36.     topic: test5

生产者需要调用生成数据

  1. package com.example.sms.middleware.sms_middleware.kafka;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.beans.factory.annotation.Value;
  5. import org.springframework.boot.ApplicationArguments;
  6. import org.springframework.boot.ApplicationRunner;
  7. import org.springframework.kafka.core.KafkaTemplate;
  8. import org.springframework.stereotype.Component;
  10. /**
  11.  * @Author kang
  12.  * @Date 2020/6/19$ 10:03$
  13.  **/
  14. @Component
  15. public class Producer {
  17.     @Autowired
  18.     private KafkaTemplate<String, String> kafkaTemplate;
  20.     @Value("${spring.kafka.topic}")
  21.     private String topic;
  23.     public void sendMessage(){
  24.         kafkaTemplate.send(topic,"message");
  25.     }
  27. //    @Override
  28. //    public void run(ApplicationArguments args) throws Exception {
  29. //        System.out.println("11111");
  30.         new Producer().sendMessage();
  31. //    }
  32. }

消费者 

  1. package com.example.sms.middleware.sms_middleware.kafka;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.springframework.kafka.annotation.KafkaListener;
  6. import org.springframework.kafka.support.Acknowledgment;
  7. import org.springframework.stereotype.Component;
  9. /**
  10.  * @Author kang
  11.  * @Date 2020/6/19$ 10:04$
  12.  **/
  13. @Component
  14. @Slf4j
  15. public class Consumer {
  17.     @KafkaListener(topics = "test5")  // 支持监听多个topic的消息
  18.     public void consumerMessage(ConsumerRecord<String, String> consumerRecord, Acknowledgment ack) {
  19.         try {
  20.             System.out.println("1212121");
  21.             String value = consumerRecord.value();
  22.             System.out.println("1212121"+value);
  23.             log.info("监听到的消息为：{}", value);
  24.             // 业务处理......
  25.         } catch (Exception e) {
  26.             e.printStackTrace();
  27.         } finally {
  28.             ack.acknowledge();
  29.         }
  30.     }
  31. }

可以直接在application.properties中添加spring.kafka.listener.concurrency=3，然后使用@KafkaListener并发消费。

对于只有一个分区的topic，不需要分区消费，因为没有意义。下面的例子是针对有2个分区的情况（我的中有4个listenPartitionX方法，我的topic设置了4个分区）

  1. @KafkaListener(id = "id0", topicPartitions = { @TopicPartition(topic = TPOIC, partitions = { "0" }) })
  2.     public void listenPartition0(List<ConsumerRecord<?, ?>> records) {
  3.         log.info("Id0 Listener, Thread ID: " + Thread.currentThread().getId());
  4.         log.info("Id0 records size " +  records.size());
  6.         for (ConsumerRecord<?, ?> record : records) {
  7.            String value = record.value();
  9.             }
  10.         }
  11.     }
  13.     @KafkaListener(id = "id1", topicPartitions = { @TopicPartition(topic = TPOIC, partitions = { "1" }) })
  14.     public void listenPartition1(List<ConsumerRecord<?, ?>> records) {
  15.         log.info("Id1 Listener, Thread ID: " + Thread.currentThread().getId());
  16.         log.info("Id1 records size " +  records.size());
  18.         for (ConsumerRecord<?, ?> record : records) {
  19.              String value = record.value();
  20.             }
  21.         }

错误接口访问不到时报错404

解决：

  1. @SpringBootApplication默认的扫描位置就是Application所在的同级目录和子目录，我们修改一下

发表评论

表情：
评论列表 （有 0 条评论，368人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Spring Boot整合Kafka

    Kafka是一个分布式的、可分区的、可复制的消息系统，下面是Kafka的几个基本术语： 1. Kafka将消息以topic为单位进行归纳； 2. 将向Kafka topi

    Bertha 。Bertha 。/ 0 赞/ 89 阅读