Apache Kafka-SpringBoot整合Kafka发送复杂对象

墨蓝 2022-10-30 05:22 257阅读 0赞

### 文章目录 ###

*  Spring Kafka概述
 *  Code
 *   *  依赖
     *  配置文件
     *  消息
     *  生产者
     *  消费者
     *  单元测试： 同步发送
     *  测试
     *  单元测试： 异步发送
 *  源码地址

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

--------------------

# Spring Kafka概述 #

Spring提供了 Spring-Kafka 项目来操作 Kafka。

[https://spring.io/projects/spring-kafka][https_spring.io_projects_spring-kafka]

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

# Code #

我们先对 Kafka-Spring 做个快速入门，实现 Producer发送消息 ，同时Consumer 消费消息。

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

--------------------

## 依赖 ##

<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.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-test</artifactId>
    			<scope>test</scope>
    		</dependency>
    		<dependency>
    			<groupId>junit</groupId>
    			<artifactId>junit</artifactId>
    			<scope>test</scope>
    		</dependency>
    	</dependencies>

Spring Boot 已经提供了 Kafka 的自动化配置的支持，但没有提供 `spring-boot-kafka-starter` 包…

![在这里插入图片描述][202102180005223.gif_pic_center]

--------------------

## 配置文件 ##

spring:
      # Kafka 配置项，对应 KafkaProperties 配置类
      kafka:
        bootstrap-servers: 192.168.126.140:9092 # 指定 Kafka Broker 地址，可以设置多个，以逗号分隔
        # Kafka Producer 配置项
        producer:
          acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。
          retries: 3 # 发送失败时，重试发送的次数
          key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化
          value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化
        # Kafka Consumer 配置项
        consumer:
          auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
          properties:
            spring:
              json:
                trusted:
                  packages: com.artisan.springkafka.domain
        # Kafka Consumer Listener 监听器配置
        listener:
          missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错
    
    logging:
      level:
        org:
          springframework:
            kafka: ERROR # spring-kafka 
          apache:
            kafka: ERROR # kafka

spring.kafka 配置项， 对应 KafkaProperties 配置类 。Spring Boot 提供的 `KafkaAutoConfiguration` 自动化配置类，实现 Kafka 的自动配置，创建相应的 Producer 和 Consumer 。

特别说明一下： 生产者 的value-serializer 配置了 Spring-Kafka 提供的 JsonSerializer 序列化类， 使用 JSON 的方式，序列化复杂的 Message 消息。

消费者的 value-serializer 配置，同样使用了 JsonDeserializer 反序列化类，因为稍后我们要使用 JSON 的方式，反序列化复杂的 Message 消息。

`properties.spring.json.trusted.packages` 需要配置`com.artisan.springkafka.domain` 包下的 Message 类们。因为 JsonDeserializer 在反序列化消息时，考虑到安全性，只反序列化成信任的 Message 类。 **务必配置**

在序列化时，使用了 JsonSerializer 序列化 Message 消息对象，它会在 Kafka 消息 Headers 的 **TypeId** 上，值为 Message 消息对应的类全名。

在反序列化时，使用了 JsonDeserializer 序列化出 Message 消息对象，它会根据 Kafka 消息 Headers 的 **TypeId** 的值，反序列化消息内容成该 Message 对象。

--------------------

## 消息 ##

package com.artisan.springkafka.domain;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/2/17 22:27 * @mark: show me the code , change the world */
    public class MessageMock { 
    
        private Integer id ;
        private String name ;
    
        public MessageMock() { 
        }
    
        public MessageMock(Integer id, String name) { 
            this.id = id;
            this.name = name;
        }
    
        public Integer getId() { 
            return id;
        }
    
        public void setId(Integer id) { 
            this.id = id;
        }
    
        public String getName() { 
            return name;
        }
    
        public void setName(String name) { 
            this.name = name;
        }
    
    
        @Override
        public String toString() { 
            return "MessageMock{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    '}';
        }
    }

--------------------

## 生产者 ##

package com.artisan.springkafka.producer;
    
    import com.artisan.springkafka.constants.TOPIC;
    import com.artisan.springkafka.domain.MessageMock;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.kafka.support.SendResult;
    import org.springframework.stereotype.Component;
    
    import java.util.Random;
    import java.util.concurrent.ExecutionException;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/2/17 22:25 * @mark: show me the code , change the world */
    
    @Component
    public class ArtisanProducerMock { 
    
    
        @Autowired
        private KafkaTemplate<Object,Object> kafkaTemplate ;
    
    
        /** * 同步发送 * @return * @throws ExecutionException * @throws InterruptedException */
        public SendResult sendMsgSync() throws ExecutionException, InterruptedException { 
            // 模拟发送的消息
            Integer id = new Random().nextInt(100);
            MessageMock messageMock = new MessageMock(id,"artisanTestMessage-" + id);
            // 同步等待
           return  kafkaTemplate.send(TOPIC.TOPIC, messageMock).get();
        }
    
    
    }

## 消费者 ##

**模拟两个不同消费组下的消费者 ，测试消费情况**

【消费者A 】

package com.artisan.springkafka.consumer;
    
    import com.artisan.springkafka.domain.MessageMock;
    import com.artisan.springkafka.constants.TOPIC;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/2/17 22:33 * @mark: show me the code , change the world */
    
    @Component
    public class ArtisanCosumerMock { 
    
    
        private Logger logger = LoggerFactory.getLogger(getClass());
        private static final String CONSUMER_GROUP_PREFIX = "MOCK-A" ;
    
        @KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC)
        public void onMessage(MessageMock messageMock){ 
            logger.info("【接受到消息][线程:{} 消息内容：{}]", Thread.currentThread().getName(), messageMock);
        }
    
    }

在方法上添加了 `@KafkaListener` 注解，指定了消费的 Topic 和 消费者分组 。

建议：建 一个类，对应一个方法。一个消费者分组，仅消费一个 Topic ，确保每个消费者分组职责单一。

--------------------

【消费者B 】

package com.artisan.springkafka.consumer;
    
    import com.artisan.springkafka.domain.MessageMock;
    import com.artisan.springkafka.constants.TOPIC;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/2/17 22:33 * @mark: show me the code , change the world */
    
    @Component
    public class ArtisanCosumerMockDiffConsumeGroup { 
    
    
        private Logger logger = LoggerFactory.getLogger(getClass());
    
        private static final String CONSUMER_GROUP_PREFIX = "MOCK-B" ;
    
        @KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC)
        public void onMessage(MessageMock messageMock){ 
            logger.info("【接受到消息][线程:{} 消息内容：{}]", Thread.currentThread().getName(), messageMock);
        }
    
    }

在方法上添加了 `@KafkaListener` 注解，指定了消费的 Topic 和 消费者分组 。

消费组和第一个消费者属于不同的消费组，请注意。

--------------------

## 单元测试： 同步发送 ##

package com.artisan.springkafka.produceTest;
    
    import com.artisan.springkafka.SpringkafkaApplication;
    import com.artisan.springkafka.producer.ArtisanProducerMock;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.kafka.support.SendResult;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutionException;
    
    /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/2/17 22:40 * @mark: show me the code , change the world */
    
    @RunWith(SpringRunner.class)
    @SpringBootTest(classes = SpringkafkaApplication.class)
    public class ProduceMockTest { 
    
        private Logger logger = LoggerFactory.getLogger(getClass());
    
    
        @Autowired
        private ArtisanProducerMock artisanProducerMock;
    
        @Test
        public void testSyncSend() throws ExecutionException, InterruptedException { 
            SendResult sendResult = artisanProducerMock.sendMsgSync();
    
            logger.info("testSyncSend Result = topic:[{}] , partition:[{}], offset:[{}]",
                    sendResult.getRecordMetadata().topic(),
                    sendResult.getRecordMetadata().partition(),
                    sendResult.getRecordMetadata().offset());
    
            // 阻塞等待，保证消费
            new CountDownLatch(1).await();
        }
    
    }

在方法内部，调用 `KafkaTemplate#send(topic, data)` 方法，异步发送消息。不过，因为后面调用了 `ListenableFuture#get()` 方法，阻塞等待发送结果，实现了同步的效果。

--------------------

## 测试 ##

运行上面的单元测试 ，

2021-02-18 00:13:50.789  INFO 24768 --- [           main] c.a.s.produceTest.ProduceMockTest        : testSyncSend Result =  topic:[MOCK_TOPIC] , partition:[0], offset:[12]
    2021-02-18 00:13:50.849  INFO 24768 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息内容：MessageMock{ id=51, name='artisanTestMessage-51'}]
    2021-02-18 00:13:50.849  INFO 24768 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock        : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{ id=51, name='artisanTestMessage-51'}]

可以看到我们发送了一个消息到`MOCK_TOPIC`上， 两个消费者属于不同的消费者组，均订阅了该TOPIC， 从结果上可以看到 该消息 可以分别被消费者组 “MOCK-ATOPIC” 和消费者组 “MOCK-BTOPIC” 都消费一次。

--------------------

但是，如果我们启动多个该示例的实例，则消费者分组 “MOCK-ATOPIC” 和 “MOCK-BTOPIC” 都会有多个 Consumer实例， 结果会怎样呢？

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70 2]

我们再运行一个， 上次的不要关哈

此时，我们再发送一条消息到`MOCK_TOPIC`，只会被 "MOCK-ATOPIC"的**一个** Consumer 消费一次，也同样只会被 “MOCK-BTOPIC” 的**一个** Consumer 消费一次。

--------------------

这个有啥用呢？ 举个例子

通过集群消费的机制，可以实现针对相同 Topic ，不同消费者分组实现各自的业务逻辑。

比如说用户注册成功时，发送一条 Topic 为 “XXXX” 的消息。 不同模块使用不同的消费者分组，订阅该 Topic ，实现各自的拓展逻辑：

*  积分模块：给用户增加 积分
 *  优惠劵模块：发放新用户专享优惠
 *  …

这样，就可以将注册成功后的业务拓展逻辑，实现业务上的解耦，未来也更加容易拓展。同时，也提高了注册接口的性能。

--------------------

## 单元测试： 异步发送 ##

com.artisan.springkafka.producer.ArtisanProducerMock

public ListenableFuture<SendResult<Object, Object>> sendMsgASync() throws ExecutionException, InterruptedException { 
            // 模拟发送的消息
            Integer id = new Random().nextInt(100);
            MessageMock messageMock = new MessageMock(id,"messageSendByAsync-" + id);
            // 异步发送消息
            ListenableFuture<SendResult<Object, Object>> result = kafkaTemplate.send(TOPIC.TOPIC, messageMock);
            return result ;
    
        }

单元测试

com.artisan.springkafka.produceTest.ProduceMockTest

@Test
        public void testAsynSend() throws ExecutionException, InterruptedException { 
                artisanProducerMock.sendMsgASync().addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() { 
                    @Override
                    public void onFailure(Throwable throwable) { 
                        logger.info(" 发送异常{}]]", throwable);
    
                    }
                    @Override
                    public void onSuccess(SendResult<Object, Object> objectObjectSendResult) { 
                        logger.info("回调结果 Result = topic:[{}] , partition:[{}], offset:[{}]",
                                objectObjectSendResult.getRecordMetadata().topic(),
                                objectObjectSendResult.getRecordMetadata().partition(),
                                objectObjectSendResult.getRecordMetadata().offset());
                    }
                });
    
            // 阻塞等待，保证消费
            new CountDownLatch(1).await();
    
        }

测试结果 同上

2021-02-18 00:40:22.443  INFO 24056 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest        : 回调结果 Result =  topic:[MOCK_TOPIC] , partition:[0], offset:[17]
    2021-02-18 00:40:22.504  INFO 24056 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock        : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{ id=37, name='messageSendByAsync-37'}]
    2021-02-18 00:40:22.504  INFO 24056 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息内容：MessageMock{ id=37, name='messageSendByAsync-37'}]

*  通过日志可以看到，发送的消息，分别被 ArtisanCosumerMock 和 ArtisanCosumerMockDiffConsumeGroup 两个消费者（位于不同的消费者分组）均消费了一次。
 *  两个消费者在不同的线程中，消费了该条消息

![在这里插入图片描述][20210218004442822.gif_pic_center]

# 源码地址 #

[https://github.com/yangshangwei/boot2/tree/master/springkafka][https_github.com_yangshangwei_boot2_tree_master_springkafka]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9hcnRpc2FuLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]: /images/20221024/4b2aa3e1913b45dbb4b22617a75c95c7.png
[https_spring.io_projects_spring-kafka]: https://spring.io/projects/spring-kafka
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70]: /images/20221024/572daddb55614381b11bc271398e585a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70 1]: /images/20221024/6cbe0a4b44ad490c8aca53dd4c1ec164.png
[202102180005223.gif_pic_center]: /images/20221024/4045855bc75642c7b5af3b492ba329f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGFuZ3dlaQ_size_16_color_FFFFFF_t_70 2]: /images/20221024/a0e1099275d4433ca3380bff86b1019c.png
[20210218004442822.gif_pic_center]: /images/20221024/dbaae6773b494e10813ae9ff4ebf5150.png
[https_github.com_yangshangwei_boot2_tree_master_springkafka]: https://github.com/yangshangwei/boot2/tree/master/springkafka