spring boot2 连接kafka

比眉伴天荒 2023-07-24 08:09 18阅读 0赞

创建项目时引入kafka

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hhdmVxaW5n_size_16_color_FFFFFF_t_70][]

项目结构

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hhdmVxaW5n_size_16_color_FFFFFF_t_70 1][]

pom.xml

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

application.yml简配版，因为是单机kafka集群，所以端口不一样了

spring:
      kafka:
        bootstrap-servers: 192.168.0.201:9092,192.168.0.201:9093,192.168.0.201:9094

复杂点的application.yml

spring:
      kafka:
        # 以逗号分隔的地址列表，用于建立与Kafka集群的初始连接(kafka 默认的端口号为9092)
        bootstrap-servers: 192.168.0.201:9092,192.168.0.201:9093,192.168.0.201:9094
        producer:
          # 发生错误后，消息重发的次数。
          retries: 0
          #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
          batch-size: 16384
          # 设置生产者内存缓冲区的大小。
          buffer-memory: 33554432
          # 键的序列化方式
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          # 值的序列化方式
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
          # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
          # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
          # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
          acks: 1
        consumer:
          # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
          auto-commit-interval: 1S
          # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
          # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
          # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
          auto-offset-reset: earliest
          # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
          enable-auto-commit: true
          # 键的反序列化方式
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          # 值的反序列化方式
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        listener:
          # 在侦听器容器中运行的线程数。
          concurrency: 5

KafkaConfig.java，非必须

package com.xin.xpfs.springbootkafka.config;
    
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class KafkaConfig {
    
        //通过KafkaAdminClient 可以在线管理kafka，创建Topic，删除Topic，查询信息等等
    
    //    @Bean
    //    public NewTopic test(){
    //        //创建topic
    //        return new NewTopic("test",1, (short) 1);
    //    }
    }

可以在kafka提前创建好topic

ProducerController.java，测试producer

package com.xin.xpfs.springbootkafka.controller;
    
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    import java.util.UUID;
    
    @RestController
    public class ProducerController {
    
        @Resource
        private KafkaTemplate<String, Object> kafkaTemplate;
    
        @RequestMapping("/send")
        public void send() {
            kafkaTemplate.send("test", UUID.randomUUID().toString());
            kafkaTemplate.send("test2", UUID.randomUUID().toString());
        }
    
    }

MyConsumer.java，listen2和listen3测试同一分组，只有一个consumer收到消息

package com.xin.xpfs.springbootkafka.consumer;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class MyConsumer {
    
        @KafkaListener(topics = "test", groupId = "group1")
        public void listen(ConsumerRecord<?,String> record) {
            String value = record.value();
            System.out.println("kafka监听的值是1："+value);
            System.out.println(record);
        }
    
        @KafkaListener(topics = "test2", groupId = "group2")
        public void listen2(ConsumerRecord<?,String> record) {
            String value = record.value();
            System.out.println("kafka监听的值是2："+value);
            System.out.println(record);
        }
    
        @KafkaListener(topics = "test2", groupId = "group2")
        public void listen3(ConsumerRecord<?,String> record) {
            String value = record.value();
            System.out.println("kafka监听的值是3："+value);
            System.out.println(record);
        }
    
    }

测试

在浏览器输出[http://localhost:8080/send][http_localhost_8080_send]

控制台输出：

kafka监听的值是1：4dc3565a-1fd9-48ba-9b05-c96ebf02755f  
kafka监听的值是2：cec42192-e704-4949-82e3-1f1044421a8e

参考：

英文文档

[https://docs.spring.io/spring-boot/docs/2.2.6.RELEASE/reference/htmlsingle/\#boot-features-kafka][https_docs.spring.io_spring-boot_docs_2.2.6.RELEASE_reference_htmlsingle_boot-features-kafka]

中文文档

[https://docshome.gitbooks.io/springboot/content/pages/spring-boot-features.html\#boot-features-kafka][https_docshome.gitbooks.io_springboot_content_pages_spring-boot-features.html_boot-features-kafka]

spring boot 2.x 系列 —— spring boot 整合 kafka

[https://blog.csdn.net/m0\_37809146/article/details/86680328][https_blog.csdn.net_m0_37809146_article_details_86680328]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hhdmVxaW5n_size_16_color_FFFFFF_t_70]: /images/20230528/8ece15ab7ee640cc8298bae5ae93f554.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hhdmVxaW5n_size_16_color_FFFFFF_t_70 1]: /images/20230528/ef12a88df50a44289c8c2bf6cefeaf4e.png
[http_localhost_8080_send]: http://localhost:8080/send
[https_docs.spring.io_spring-boot_docs_2.2.6.RELEASE_reference_htmlsingle_boot-features-kafka]: https://docs.spring.io/spring-boot/docs/2.2.6.RELEASE/reference/htmlsingle/#boot-features-kafka
[https_docshome.gitbooks.io_springboot_content_pages_spring-boot-features.html_boot-features-kafka]: https://docshome.gitbooks.io/springboot/content/pages/spring-boot-features.html#boot-features-kafka
[https_blog.csdn.net_m0_37809146_article_details_86680328]: https://blog.csdn.net/m0_37809146/article/details/86680328