springboot 与 kafka整合示例

傷城~ 2023-07-02 11:30 120阅读 0赞

## kafka 是什么？ ##

Kafka是由Apache软件基金会开发的一个**开源流处理平台**，由Scala和Java编写。该项目的目标是**为处理实时数据提供一个统一、高吞吐、低延迟的平台**。**其持久化层本质上是一个“按照分布式事务日志架构的大规模发布/订阅消息队列”**，这使它作为企业级基础设施来处理流式数据非常有价值。此外，Kafka可以通过Kafka Connect连接到外部系统（用于数据输入/输出），并提供了Kafka Streams——一个Java流式处理库。

1.  可以进行消息发布和订阅
2.  可以存储数据
3.  可以处理实时流数据

kafka基本概念

*  **Producer：消息和数据的生产者，向Kafka的一个topic发布消息的进程/代码/服务**
 *  **Consumer：消息和数据的消费者，订阅数据(Topic)并且处理其发布的消息的进程/代码/服务**
 *  **Consumer Group:逻辑概念，对于同一个topic,会广播给不同的group，一个group中，只有一个consumer可以消费该消息**
 *  **Broker：物理概念，Kafka集群中的每个Kafka节点**
 *  **Topic：逻辑概念，Kafka消息的类别，对数据进行区分、隔离**
 *  **Partition：物理概念，Kafka下数据存储的基本单元。一个Topic数据，会被分散存储到多个Partition，每一个Partition是有序的**
 *  **Replication：同一个Partition可能会有多个Replica，多个Replica之间数据是一样的**
 *  **Replication Leader：一个Partition的多个Replica上，需要一个Leader负责该Partition上与Producer和Consumer交互**
 *  **ReplicaManager：负责管理当前broker所有分区和副本的信息、处理KafkaController发起的一些请求，副本状态的切换、添加/读取消息等**

## 创建项目 ##

jdk1.8 、springboot 、fastjson、lombok

### 项目整体架构： ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdqaWUxNjE2_size_16_color_FFFFFF_t_70]  
本文使用IDEA 创建 springboot 项目，简单快速，kafka服务搭建此处不详细讲解,有兴趣可以查看我的这篇博文 [《在windows上搭建 kafka》][windows_ kafka]

### 1. 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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.2.4.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.example</groupId>
        <artifactId>springboot-kafka-demo</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>springboot-kafka-demo</name>
        <description>Demo project for Spring Boot</description>
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.62</version>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
                <exclusions>
                    <exclusion>
                        <groupId>org.junit.vintage</groupId>
                        <artifactId>junit-vintage-engine</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

### 2. 配置 application.yml 文件 ###

server:
      port: 8080
    spring:
      application:
        name: springboot-kafka-demo
      kafka: # 指定kafka的代理地址，可以配置多个
        bootstrap-servers: localhost:9092,localhost:9093,localhost:9094
        producer:
          retries: 0 # 每次批量发送消息的数量
          # 这有助于提升客户端和服务器上的性能，此配置控制默认批量大小（以字节为单位），默认值为16384
          batch-size: 16384
          # 生产者可用于缓冲等待发送到服务器的记录的内存总字节数，默认值为33554432
          buffer-memory: 33554432
          # key的Serializer类，实现类实现了接口org.apache.kafka.common.serialization.Serializer
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          # value的Serializer类，实现类实现了接口org.apache.kafka.common.serialization.Serializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
          bootstrap-servers: ${ spring.kafka.bootstrap-servers}
        consumer:
          bootstrap-servers: ${ spring.kafka.bootstrap-servers}
          # 指定默认消费者group-id
          group-id: test
          # earliest 如果不存在已提交的offest时，表示从头开始消费
          auto-offset-reset: earliest
          # 默认采用自动提交的机制
          enable-auto-commit: true
          # 如果'enable.auto.commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。
          auto-commit-interval: 100
          # 指定消息key和消息体的编解码方式
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

### 3. 生产者Producer ###

package com.example.springboot.kafka.demo.kafka.producer;
    
    import com.alibaba.fastjson.JSON;
    import com.example.springboot.kafka.demo.kafka.Message;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Component;
    import org.springframework.web.bind.annotation.RequestBody;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    /** * @version 1.0.0 * @Description 生产者测试 * @createTime 2020年01月29日 11:59:00 */
    @Component
    @Slf4j
    @RestController
    public class KafkaSender { 
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        /** * 发送消息方法 * @param msg */
        @RequestMapping("/send")
        public void send(@RequestBody Message msg){ 
            log.info("message = {}", JSON.toJSONString(msg));
            kafkaTemplate.send("test",JSON.toJSONString(msg));
        }
    }

### 4. 消息监听 Consumer ###

package com.example.springboot.kafka.demo.kafka.consumer;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    import java.util.Optional;
    
    /** * @version 1.0.0 * @Description 消费者 * @createTime 2020年01月29日 13:18:00 */
    @Slf4j
    @Component
    public class KafkaReceiver { 
        @KafkaListener(topics = "test")
        public void listen(ConsumerRecord<?,?> record){ 
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if(kafkaMessage.isPresent()){ 
                Object message = kafkaMessage.get();
                log.info("record------------record = " + record);
                log.info("message----------- message = " + message );
            }
        }
    }

### 5. 消息实体 Message ###

package com.example.springboot.kafka.demo.kafka;
    
    import lombok.Builder;
    import lombok.Data;
    
    import java.util.Date;
    
    /** * @version 1.0.0 * @Description 消息体 * @createTime 2020年01月29日 12:39:00 */
    @Data
    @Builder
    public class Message { 
        private Long id;
        private String msg;
        private Date sendTime;
    }

### 6 测试 ###

#### 6.1 启动项目 ####

#### 6.2 使用postman进行测试 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdqaWUxNjE2_size_16_color_FFFFFF_t_70 1]

#### 6.3 控制台打印 ####

![在这里插入图片描述][20200129222651969.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdqaWUxNjE2_size_16_color_FFFFFF_t_70 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdqaWUxNjE2_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200129211358349.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdqaWUxNjE2,size_16,color_FFFFFF,t_70
[windows_ kafka]: https://blog.csdn.net/wangjie1616/article/details/104111418
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdqaWUxNjE2_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200129222543351.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdqaWUxNjE2,size_16,color_FFFFFF,t_70
[20200129222651969.png]: https://img-blog.csdnimg.cn/20200129222651969.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdqaWUxNjE2_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200129222813392.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdqaWUxNjE2,size_16,color_FFFFFF,t_70