SpringBoot（十二）：SpringBoot整合Kafka

今天药忘吃喽~ 2022-05-27 08:49 329阅读 0赞

#### 一、准备工作 ####

> 提前说明：如果你运行出问题，请检查Kafka的版本与SpringBoot的版本是否与我文中的一致，本文中的环境已经经过测试。

*  Kafka服务版本为 kafka\_2.11-1.1.0 (Scala)， 也就是1.1.0
 *  SpringBoot版本：1.5.10.RELEASE
 *  提前启动zk，kafka，并且创建一个Topic

[root@Basic kafka_2.11-1.1.0]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test_topic

*  确保你的kafka能够访问，如果访问不了，需要打开外网访问。

`config/server.properties`

advertised.listeners=PLAINTEXT://192.168.239.128:9092

*  Maven 依赖

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

#### 二、项目结构 ####

> 为了更加体现实际开发需求，一般生产者都是在调用某些接口的服务处理完逻辑之后然后往kafka里面扔数据，然后有一个消费者不停的监控这个Topic，然后处理数据，所以这里把生产者作为一个接口，消费者放到kafka这个目录下，注意@Component注解，不然扫描不到@KafkaListener

![这里写图片描述][70]

#### 三、具体实现代码 ####

##### SpringBoot配置文件 #####

application.yml

spring:
      kafka:
        bootstrap-servers: 192.168.239.128:9092
        producer:
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        consumer:
          group-id: test
          enable-auto-commit: true
          auto-commit-interval: 1000
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

##### 生产者 #####

package cn.saytime.web;
    
    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 TestKafkaProducerController { 
    
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
        @RequestMapping("send")
        public String send(String msg){
            kafkaTemplate.send("test_topic", msg);
            return "success";
        }
    
    }

##### 消费者 #####

> 这里的消费者会监听这个主题，有消息就会执行，不需要进行while(true)

package cn.saytime.kafka;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    /** * kafka消费者测试 */
    @Component
    public class TestConsumer { 
    
        @KafkaListener(topics = "test_topic")
        public void listen (ConsumerRecord<?, ?> record) throws Exception {
            System.out.printf("topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
        }
    }

项目启动类

package cn.saytime;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class TestApplication{ 
    
        public static void main(String[] args) {
            SpringApplication.run(TestApplication.class, args);
        }
    }

#### 四、测试 ####

运行项目，执行：[http://localhost:8080/kafka/send?msg=hello][http_localhost_8080_kafka_send_msg_hello]

控制台输出：

topic = test_topic, offset = 19, value = hello

为了体现消费者不止执行一次就结束，再调用一次接口：  
[http://localhost:8080/kafka/send?msg=kafka][http_localhost_8080_kafka_send_msg_kafka]

topic = test_topic, offset = 20, value = kafka

> 所以可以看到这里消费者实际上是不停的poll Topic数据的。

[70]: /images/20220527/0d15d1d38ba3463ab984193fd4325661.png
[http_localhost_8080_kafka_send_msg_hello]: http://localhost:8080/kafka/send?msg=hello
[http_localhost_8080_kafka_send_msg_kafka]: http://localhost:8080/kafka/send?msg=kafka