快速搭建Springboot集成apache Kafka环境，测试示例

布满荆棘的人生 2024-04-18 11:19 55阅读 0赞

### 目录 ###

*   *  一、Java环境
     *  二、kafka环境
     *  三、代码示例,项目结构
     *  四、更多Kafka,，客户端、服务端，监控可参考下文

## 一、Java环境 ##

## 二、kafka环境 ##

基于docker 安装，开启服务，对外端口映射即可，创建Topic 使用Java Api操作即可。可参考其他资料关于docker/kafka的安装。  
这是kafka的目录信息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5_size_16_color_FFFFFF_t_70]

## 三、代码示例,项目结构 ##

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

maven + Java8 + IntelliJ IDEA + Springboot

*  3.1 创建Maven工程，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>com.tonels</groupId>
    			<artifactId>common</artifactId>
    			<version>1.0-SNAPSHOT</version>
    		</dependency>
    
    		<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.2 Springboot集成 kafka,通过Yml文件

server:
      servlet:
        context-path: /
      port: 8080
    spring:
      kafka:
        bootstrap-servers: 192.168.1.74: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: ls

*  3.3 模拟生产者  
    消息体message数据封装

@Data
    public class Message { 
        private String id;
        private String msg;
        private LocalDateTime sendTime;
    }

生产过程

@Component
    public class Producer { 
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        private static Gson gson = new GsonBuilder().create();
    
        //发送消息方法
        public void send() { 
            Message message = new Message();
            message.setId("KF_"+System.currentTimeMillis());
            message.setMsg(UUID.randomUUID().toString());
            message.setSendTime(LocalDateTime.now());
    
            // 指定topic sl_test
            kafkaTemplate.send("sl_test",JSONUtil.toJsonStr(message));
        }
    }
    ```java
    
    发送消息,模拟http发送消息
    ```java
    @RestController
    @RequestMapping("/kafka")
    public class SendController { 
    
        @Autowired
        private Producer producer;
    
        @RequestMapping(value = "/send")
        public String send() { 
            producer.send();
            return "{\"code\":0}";
        }
    }

*  3.4 消费者自动消费Topic,并相应业务处理

@Component
    public class Consumer { 
    
        @KafkaListener(topics = { "sl_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);
    
                Message m = JSONUtil.toBean(JSONUtil.toJsonStr(message), Message.class);
                System.out.println("转成对象后。。" + m);
            }
    
        }
    }

*  3.5 控制台输出

---->ConsumerRecord(topic = sl_test, partition = 0, offset = 29, CreateTime = 1567696265396, serialized key size = -1, serialized value size = 107, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = {"msg":"2689bc7a-9781-45ab-80b1-2b54b94c872b","id":"KF_1567696265344","sendTime":"2019-09-05T23:11:05.348"})
    ---->{"msg":"2689bc7a-9781-45ab-80b1-2b54b94c872b","id":"KF_1567696265344","sendTime":"2019-09-05T23:11:05.348"}
    转成对象后。。Message(id=KF_1567696265344, msg=2689bc7a-9781-45ab-80b1-2b54b94c872b, sendTime=2019-09-05T23:11:05.348)

*  3.6 查看服务器日志信息  
    首先会自动创建 Topic  
    ![在这里插入图片描述][20190906000052734.png]  
    日志文件里会基于生产的消息，存在日志  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5_size_16_color_FFFFFF_t_70 2]

## 四、更多Kafka,，客户端、服务端，监控可参考下文 ##

*  [kafka学习 – kafka connect][kafka_ _ kafka connect]
 *  [kafka学习 – 客户端Consumer API学习][kafka_ _ _Consumer API]
 *  [kafka学习 – 客户端Producer API学习][kafka_ _ _Producer API]
 *  [kafka学习 – 服务端命令学习][kafka_ _]
 *  [Springboot集成Kafka][Springboot_Kafka]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/5a2a8aaca22845519eba04ec497ae39c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/3d69247da5144c46b94b08c34c526d15.png
[20190906000052734.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/1932443c0daf4ad19ac69ecbc05218d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/3f457f3c91074b78992bfa3f1bdce8cc.png
[kafka_ _ kafka connect]: https://blog.csdn.net/qq_42105629/article/details/107077881
[kafka_ _ _Consumer API]: https://blog.csdn.net/qq_42105629/article/details/107073062
[kafka_ _ _Producer API]: https://blog.csdn.net/qq_42105629/article/details/107072983
[kafka_ _]: https://blog.csdn.net/qq_42105629/article/details/107008616
[Springboot_Kafka]: https://blog.csdn.net/qq_42105629/article/details/100569364