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

---

Spring Kafka概述

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

https://spring.io/projects/spring-kafka

Code

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

---

依赖

代码语言：javascript复制

	<dependencies>
		<dependency>
			<groupId>org.springframework.bootgroupId>
			<artifactId>spring-boot-starter-webartifactId>
		dependency>

		 
		<dependency>
			<groupId>org.springframework.kafkagroupId>
			<artifactId>spring-kafkaartifactId>
		dependency>

		<dependency>
			<groupId>org.springframework.bootgroupId>
			<artifactId>spring-boot-starter-testartifactId>
			<scope>testscope>
		dependency>
		<dependency>
			<groupId>junitgroupId>
			<artifactId>junitartifactId>
			<scope>testscope>
		dependency>
	dependencies>

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

---

配置文件

代码语言：javascript复制

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 对象。

---

消息

代码语言：javascript复制

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   '''  
                '}';
    }
}

---

生产者

代码语言：javascript复制

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 】

代码语言：javascript复制

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 】

代码语言：javascript复制

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 和 消费者分组 。

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

---

单元测试： 同步发送

代码语言：javascript复制

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() 方法，阻塞等待发送结果，实现了同步的效果。

---

测试

运行上面的单元测试 ，

代码语言：javascript复制

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实例， 结果会怎样呢？

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

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

---

这个有啥用呢？ 举个例子

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

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

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

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

---

单元测试： 异步发送

com.artisan.springkafka.producer.ArtisanProducerMock

代码语言：javascript复制

  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

代码语言：javascript复制

    @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();

    }

测试结果 同上

代码语言：javascript复制

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 两个消费者（位于不同的消费者分组）均消费了一次。
• 两个消费者在不同的线程中，消费了该条消息

源码地址

https://github.com/yangshangwei/boot2/tree/master/springkafka

文件存储 kafka 单元测试 spring springboot

0 人点赞