springboot集成kafka消息

喜欢ヅ旅行 2023-08-17 17:27 65阅读 0赞

kafka的安装见 [https://blog.csdn.net/qq\_33371766/article/details/87866533][https_blog.csdn.net_qq_33371766_article_details_87866533]

1.创建springboot项目

2.pom.xml

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
       <modelVersion>4.0.0</modelVersion>
    
       <groupId>com.xiaour.spring.boot</groupId>
       <artifactId>kafka</artifactId>
       <version>0.0.1-SNAPSHOT</version>
       <packaging>jar</packaging>
    
       <name>kafka</name>
       <description>Kafka Demo project for Spring Boot</description>
    
       <parent>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-parent</artifactId>
          <version>2.0.2.RELEASE</version>
          <relativePath/> 
       </parent>
    
       <properties>
          <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
          <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
          <java.version>1.8</java.version>
       </properties>
    
       <dependencies>
          <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter</artifactId>
          </dependency>
          <dependency>
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka</artifactId>
          </dependency>
    
          <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
          </dependency>
    
          <dependency>
             <groupId>com.google.code.gson</groupId>
             <artifactId>gson</artifactId>
             <version>2.8.2</version>
          </dependency>
          <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
             <version>RELEASE</version>
          </dependency>
    
       </dependencies>
    
       <build>
          <plugins>
             <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
             </plugin>
          </plugins>
       </build>
    
    
    </project>

3.application.yml

server:
      servlet:
        context-path: /
      port: 8080
    spring:
      kafka:
        bootstrap-servers: 127.0.0.1:9092
        #生产者的配置，大部分我们可以使用默认的，这里列出几个比较重要的属性
        producer:
          #每批次发送消息的数量
          batch-size: 16
          #设置大于0的值将使客户端重新发送任何数据，一旦这些数据发送失败。注意，这些重试与客户端接收到发送错误时的重试没有什么不同。允许重试将潜在的改变数据的顺序，如果这两个消息记录都是发送到同一个partition，则第一个消息失败第二个发送成功，则第二条消息会比第一条消息出现要早。
          retries: 0
          #producer可以用来缓存数据的内存大小。如果数据产生速度大于向broker发送的速度，producer会阻塞或者抛出异常，以“block.on.buffer.full”来表明。这项设置将和producer能够使用的总内存相关，但并不是一个硬性的限制，因为不是producer使用的所有内存都是用于缓存。一些额外的内存会用于压缩（如果引入压缩机制），同样还有一些用于维护请求。
          buffer-memory: 33554432
          #key序列化方式
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        #消费者的配置
        consumer:
          #Kafka中没有初始偏移或如果当前偏移在服务器上不再存在时,默认区最新 ，有三个选项 【latest, earliest, none】
          auto-offset-reset: latest
          #是否开启自动提交
          enable-auto-commit: true
          #自动提交的时间间隔
          auto-commit-interval: 100
          #key的解码方式
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          #value的解码方式
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          #在/usr/local/etc/kafka/consumer.properties中有配置
          group-id: test-consumer-group

4.consumer

import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    import java.util.Optional;
    
    @Component
    public class Consumer {
    
        @KafkaListener(topics = {"test"})
        public void listen(ConsumerRecord<?, ?> record){
    
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
    
            if (kafkaMessage.isPresent()) {
    
                Object message = kafkaMessage.get();
                System.out.println("---->"+record);
                System.out.println("---->"+message);
    
            }
    
        }
    }

5.1 Message

import java.util.Date;
    
    public class Message {
    
        private String id;
    
        private String msg;
    
        private Date sendTime;
    
        public String getId() {
            return id;
        }
    
        public void setId(String id) {
            this.id = id;
        }
    
        public String getMsg() {
            return msg;
        }
    
        public void setMsg(String msg) {
            this.msg = msg;
        }
    
        public Date getSendTime() {
            return sendTime;
        }
    
        public void setSendTime(Date sendTime) {
            this.sendTime = sendTime;
        }
    }

5.2 Producer

import com.google.gson.Gson;
    import com.google.gson.GsonBuilder;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
    import java.util.UUID;
    
    @Component
    public class Producer {
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        private static Gson gson = new GsonBuilder().create();
    
        //发送消息方法
        public void send() {
            Message message = new Message();
            message.setId("KFK_"+System.currentTimeMillis());
            message.setMsg(UUID.randomUUID().toString());
            message.setSendTime(new Date());
            kafkaTemplate.send("test", gson.toJson(message));
        }
    
    }

5.3 SendController

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    @RequestMapping("/kafka")
    public class SendController {
    
        @Autowired
        private Producer producer;
    
        @RequestMapping(value = "/send")
        public String send() {
            producer.send();
            return "{\"code\":0}";
        }
    }

6.启动类

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

7.测试

先启动本地的kafka和zookeeper

[http://localhost:8080/kafka/send][http_localhost_8080_kafka_send]

![20191001123501989.png][]

![2019100112352741.png][]

测试ok！！！

代码地址：[https://github.com/ChenXbFrank/SpringBootKafkaDemo][https_github.com_ChenXbFrank_SpringBootKafkaDemo]

[文章参考][Link 1]

[https_blog.csdn.net_qq_33371766_article_details_87866533]: https://blog.csdn.net/qq_33371766/article/details/87866533
[http_localhost_8080_kafka_send]: http://localhost:8080/kafka/send
[20191001123501989.png]: /images/20230808/776976e12b8d4178b94549fffda5767c.png
[2019100112352741.png]: /images/20230808/cd7c18a82f98404b8448b295a32a70d4.png
[https_github.com_ChenXbFrank_SpringBootKafkaDemo]: https://github.com/ChenXbFrank/SpringBootKafkaDemo
[Link 1]: http://www.54tianzhisheng.cn/2018/01/05/SpringBoot-Kafka/