⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 http://www.iocoder.cn/Spring-Boot/Kafka/ 「芋道源码」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注**微信公众号：【芋道源码】**有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

本文在提供完整代码示例，可见 https://github.com/YunaiV/SpringBoot-Labs 的 lab-03-kafka 目录。

原创不易，给点个 Star 嘿，一起冲鸭！

1. 概述

如果胖友还没了解过分布式消息队列 Apache Kafka ，建议先阅读下艿艿写的《芋道 Kafka 极简入门》文章。虽然这篇文章标题是安装部署，实际可以理解成《一文带你快速入门 Kafka》，哈哈哈。

考虑这是 Kafka 如何在 Spring Boot 整合与使用的文章，所以还是简单介绍下 Kafka 是什么？

FROM 《分布式发布订阅消息系统 Kafka》

Kafka 是一种高吞吐量的分布式发布订阅消息系统，她有如下特性：

通过 O(1) 的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。

高吞吐量：即使是非常普通的硬件kafka也可以支持每秒数十万的消息。

支持通过 Kafka 服务器和消费机集群来分区消息。

在本文中，我们会比《芋道 Kafka 极简入门》提供更多的生产者 Producer 和消费者 Consumer 的使用示例。例如说：

Producer 三种发送消息的方式。
Producer 发送顺序消息，Consumer 顺序消费消息。
~~Producer 发送定时消息。~~（暂不支持）
Producer 批量发送消息。
Producer 发送事务消息。
Consumer 批量消费消息。
Consumer 广播和集群消费消息。

胖友你就说，艿艿是不是很良心。😜

2. Spring-Kafka

在 Spring 生态中，提供了 Spring-Kafka 项目，让我们更简便的使用 Kafka 。其官网介绍如下：

The Spring for Apache Kafka (spring-kafka) project applies core Spring concepts to the development of Kafka-based messaging solutions.
Spring for Apache Kafka (spring-kafka) 项目将 Spring 核心概念应用于基于 Kafka 的消息传递解决方案的开发。

It provides a "template" as a high-level abstraction for sending messages.
它提供了一个“模板”作为发送消息的高级抽象。

It also provides support for Message-driven POJOs with @KafkaListener annotations and a "listener container".
它还通过 @KafkaListener 注解和“侦听器容器(listener container)”为消息驱动的 POJO 提供支持。

These libraries promote the use of dependency injection and declarative.
这些库促进了依赖注入和声明的使用。

In all of these cases, you will see similarities to the JMS support in the Spring Framework and RabbitMQ support in Spring AMQP.
在所有这些用例中，你将看到 Spring Framework 中的 JMS 支持，以及和 Spring AMQP 中的 RabbitMQ 支持的相似之处。

😈 注意，Spring-Kafka 是基于 Spring Message 来实现 Kafka 的发送端和接收端。

Features(功能特性)

KafkaTemplate

KafkaMessageListenerContainer

@KafkaListener

KafkaTransactionManager

spring-kafka-test jar with embedded kafka server(带嵌入式 Kafka 服务器的 spring-kafka-test jar 包)

3. 快速入门

示例代码对应仓库：lab-31-kafka-demo 。

本小节，我们先来对 Kafka-Spring 做一个快速入门，实现 Producer 三种发送消息的方式的功能，同时创建一个 Consumer 消费消息。

考虑到一个应用既可以使用生产者 Producer ，又可以使用消费者 Consumer ，所以示例就做成一个 lab-31-kafka-demo 项目。

3.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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>lab-03-kafka-demo</artifactId>

    <dependencies>
        <!-- 引入 Spring-Kafka 依赖 -->
        <!-- 已经内置 kafka-clients 依赖，所以无需重复引入 -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.3.3.RELEASE</version>
        </dependency>

        <!-- 实现对 JSON 的自动化配置 -->
        <!-- 因为，Kafka 对复杂对象的 Message 序列化时，我们会使用到 JSON -->
        <!--
            同时，spring-boot-starter-json 引入了 spring-boot-starter ，而 spring-boot-starter 又引入了 spring-boot-autoconfigure 。
            spring-boot-autoconfigure 实现了 Spring-Kafka 的自动化配置
         -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-json</artifactId>
        </dependency>

        <!-- 方便等会写单元测试 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

具体每个依赖的作用，胖友自己认真看下艿艿添加的所有注释噢。
不过有点很奇怪的是，Spring Boot 已经提供了 Kafka 的自动化配置的支持，但是竟然没有提供 spring-boot-kafka-starter 包，有点神奇~

3.2 应用配置文件

在 resources 目录下，创建 application.yaml 配置文件。配置如下：

spring:
  # Kafka 配置项，对应 KafkaProperties 配置类
  kafka:
    bootstrap-servers: 127.0.0.1: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 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring:
          json:
            trusted:
              packages: cn.iocoder.springboot.lab03.kafkademo.message
    # Kafka Consumer Listener 监听器配置
    listener:
      missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错

logging:
  level:
    org:
      springframework:
        kafka: ERROR # spring-kafka INFO 日志太多了，所以我们限制只打印 ERROR 级别
      apache:
        kafka: ERROR # kafka INFO 日志太多了，所以我们限制只打印 ERROR 级别

在 spring.kafka 配置项，设置 Kafka 的配置，对应 KafkaProperties 配置类。
Spring Boot 提供的 KafkaAutoConfiguration 自动化配置类，实现 Kafka 的自动配置，创建相应的 Producer 和 Consumer 。
spring.kafka.bootstrap-servers 配置项，设置 Kafka Broker 地址。如果多个，使用逗号分隔。
spring.kafka.producer 配置项，一看就知道是 Kafka Producer 所独有。
- value-serializer 配置，我们使用了 Spring-Kafka 提供的 JsonSerializer 序列化类，因为稍后我们要使用 JSON 的方式，序列化复杂的 Message 消息。
- 其它配置，一般默认即可。
spring.kafka.consumer 配置项，一看就知道是 Kafka Consumer 所独有。
- value-serializer 配置，我们使用了 Spring-Kafka 提供的 JsonDeserializer 反序列化类，因为稍后我们要使用 JSON 的方式，反序列化复杂的 Message 消息。
- properties.spring.json.trusted.packages 配置，配置信任 cn.iocoder.springboot.lab03.kafkademo.message 包下的 Message 类们。因为 JsonDeserializer 在反序列化消息时，考虑到安全性，只反序列化成信任的 Message 类。😈 想要尝试下效果的胖友，可以选择去掉这个配置，很酸爽。

3.3 Application

创建 Application.java 类，配置 @SpringBootApplication 注解即可。代码如下：

// Application.java

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

}

3.4 Demo01Message

在 cn.iocoder.springboot.lab03.kafkademo.message 包下，创建 Demo01Message 消息类，提供给当前示例使用。代码如下：

// Demo01Message.java

public class Demo01Message {

    public static final String TOPIC = "DEMO_01";

    /**
     * 编号
     */
    private Integer id;

    // ... 省略 set/get/toString 方法

}

TOPIC 静态属性，我们设置该消息类对应 Topic 为 "DEMO_01" 。

3.5 Demo01Producer

在 cn.iocoder.springboot.lab03.kafkademo.producer 包下，创建 Demo01Producer 类，它会使用 Kafka-Spring 封装提供的 KafkaTemplate ，实现三种发送消息的方式。代码如下：

// Demo01Producer.java

@Component
public class Demo01Producer {

    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;

    public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
        // 创建 Demo01Message 消息
        Demo01Message message = new Demo01Message();
        message.setId(id);
        // 同步发送消息
        return kafkaTemplate.send(Demo01Message.TOPIC, message).get();
    }

    public ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id) {
        // 创建 Demo01Message 消息
        Demo01Message message = new Demo01Message();
        message.setId(id);
        // 异步发送消息
        return kafkaTemplate.send(Demo01Message.TOPIC, message);
    }

}

#asyncSend(...) 方法，异步发送消息。在方法内部，会调用 KafkaTemplate#send(topic, data) 方法，异步发送消息，返回 Spring ListenableFuture 对象，一个可以通过监听执行结果的 Future 增强。
#syncSend(...) 方法，同步发送消息。在方法内部，也是调用 KafkaTemplate#send(topic, data) 方法，异步发送消息。不过，因为我们后面调用了 ListenableFuture 对象的 #get() 方法，阻塞等待发送结果，从而实现同步的效果。
暂时未提供 oneway 发送消息的方式。因为需要配置 Producer 的 acks = 0 ，才可以使用这种发送方式。😈 当然，实际场景下，基本不会使用 oneway 的方式来发送消息，所以直接先忽略吧。

对于胖友来说，可能最关心的是，消息 Message 是怎么序列化的。

在序列化时，我们使用了 JsonSerializer 序列化 Message 消息对象，它会在 Kafka 消息 Headers 的 __TypeId__ 上，值为 Message 消息对应的类全名。
在反序列化时，我们使用了 JsonDeserializer 序列化出 Message 消息对象，它会根据 Kafka 消息 Headers 的 __TypeId__ 的值，反序列化消息内容成该 Message 对象。

3.6 Demo01Consumer

在 cn.iocoder.springboot.lab03.kafkademo.consumer 包下，创建 Demo01Consumer 类，消费消息。代码如下：

// Demo01Consumer.java

@Component
public class Demo01Consumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = Demo01Message.TOPIC,
            groupId = "demo01-consumer-group-" + Demo01Message.TOPIC)
    public void onMessage(Demo01Message message) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

}

在方法上，添加了 @KafkaListener 注解，声明消费的 Topic 是 "DEMO_01" ，消费者分组是 "demo01-consumer-group-DEMO_01" 。一般情况下，我们建议一个消费者分组，仅消费一个 Topic 。这样做会有个好处：每个消费者分组职责单一，只消费一个 Topic 。
方法参数，使用消费 Topic 对应的消息类即可。这里，我们使用了「3.4 Demo01Message」。
虽然说，@KafkaListener 注解是方法级别的，艿艿还是建议一个类，对应一个方法，消费消息。😈 简单清晰~

3.7 Demo01AConsumer

在 cn.iocoder.springboot.lab03.kafkademo.consumer 包下，创建 Demo01AConsumer 类，消费消息。代码如下：

// Demo01AConsumer.java

@Component
public class Demo01AConsumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = Demo01Message.TOPIC,
            groupId = "demo01-A-consumer-group-" + Demo01Message.TOPIC)
    public void onMessage(ConsumerRecord<Integer, String> record) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), record);
    }
    
}

整体和「3.6 Demo01Consumer」是一致的，主要有两个差异点，也是为什么我们又额外创建了这个消费者的原因。

差异一，在方法上，添加了 @KafkaListener 注解，声明消费的 Topic 还是 "DEMO_01" ，消费者分组修改成了 "demo01-A-consumer-group-DEMO_01" 。这样，我们就可以测试 Kafka 集群消费的特性。

集群消费（Clustering）：集群消费模式下，相同 Consumer Group 的每个 Consumer 实例平均分摊消息。

也就是说，如果我们发送一条 Topic 为 "DEMO_01" 的消息，可以分别被 "demo01-A-consumer-group-DEMO_01" 和 "demo01-consumer-group-DEMO_01" 都消费一次。
但是，如果我们启动两个该示例的实例，则消费者分组 "demo01-A-consumer-group-DEMO_01" 和 "demo01-consumer-group-DEMO_01" 都会有多个 Consumer 示例。此时，我们再发送一条 Topic 为 "DEMO_01" 的消息，只会被 "demo01-A-consumer-group-DEMO_01" 的一个 Consumer 消费一次，也同样只会被 "demo01-A-consumer-group-DEMO_01" 的一个 Consumer 消费一次。

好好理解上述的两段话，非常重要。

通过集群消费的机制，我们可以实现针对相同 Topic ，不同消费者分组实现各自的业务逻辑。例如说：用户注册成功时，发送一条 Topic 为 "USER_REGISTER" 的消息。然后，不同模块使用不同的消费者分组，订阅该 Topic ，实现各自的拓展逻辑：

积分模块：判断如果是手机注册，给用户增加 20 积分。
优惠劵模块：因为是新用户，所以发放新用户专享优惠劵。
站内信模块：因为是新用户，所以发送新用户的欢迎语的站内信。
... 等等

这样，我们就可以将注册成功后的业务拓展逻辑，实现业务上的解耦，未来也更加容易拓展。同时，也提高了注册接口的性能，避免用户需要等待业务拓展逻辑执行完成后，才响应注册成功。

差异二，方法参数，设置消费的消息对应的类不是 Demo01Message 类，而是 Kafka 内置的 ConsumerRecord 类。通过 ConsumerRecord 类，我们可以获取到消费的消息的更多信息，例如说消息的所属队列、创建时间等等属性，不过消息的内容(value)就需要自己去反序列化。当然，一般情况下，我们不会使用 ConsumerRecord 类。

3.8 简单测试

创建 Demo01ProducerTest 测试类，编写二个单元测试方法，调用 Demo01Producer 二个发送消息的方式。代码如下：

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class Demo01ProducerTest {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private Demo01Producer producer;

    @Test
    public void testSyncSend() throws ExecutionException, InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        SendResult result = producer.syncSend(id);
        logger.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }

    @Test
    public void testASyncSend() throws InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        producer.asyncSend(id).addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {

            @Override
            public void onFailure(Throwable e) {
                logger.info("[testASyncSend][发送编号：[{}] 发送异常]]", id, e);
            }

            @Override
            public void onSuccess(SendResult<Object, Object> result) {
                logger.info("[testASyncSend][发送编号：[{}] 发送成功，结果为：[{}]]", id, result);
            }

        });

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }
    
}

比较简单，胖友自己看下三个单元测试方法。

我们来执行 #testSyncSend() 方法，测试同步发送消息。控制台输出如下：

# Producer 同步发送消息成功。注意 __TypeId__
2019-12-08 18:14:11.174  INFO 89529 --- [           main] c.i.s.l.k.producer.Demo01ProducerTest    : [testSyncSend][发送编号：[1575627250] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=DEMO_01, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 49, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo01Message{id=1575627250}, timestamp=null), recordMetadata=DEMO_01-0@0]]]

# Demo01AConsumer 消费了一次该消息
2019-12-08 18:14:11.217  INFO 89529 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo01AConsumer       : [onMessage][线程编号:16 消息内容：ConsumerRecord(topic = DEMO_01, partition = 0, leaderEpoch = 0, offset = 0, CreateTime = 1575627251158, serialized key size = -1, serialized value size = 17, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = Demo01Message{id=1575627250})]

# Demo01Consumer 消费了一次该消息
2019-12-08 18:14:11.220  INFO 89529 --- [ntainer#1-0-C-1] c.i.s.l.k.consumer.Demo01Consumer        : [onMessage][线程编号:18 消息内容：Demo01Message{id=1575627250}]

通过日志我们可以看到，我们发送的消息，分别被 Demo01AConsumer 和 Demo01Consumer 两个消费者（消费者分组）都消费了一次。
同时，两个消费者在不同的线程中，消费了这条消息。

我们来执行 #testASyncSend() 方法，测试异步发送消息。控制台输出如下：

友情提示：注意，不要关闭 #testSyncSend() 单元测试方法，因为我们要模拟每个消费者集群，都有多个 Consumer 节点。

// Producer 异步发送消息成功
2019-12-08 18:20:34.096  INFO 89818 --- [ad | producer-1] c.i.s.l.k.producer.Demo01ProducerTest    : [testASyncSend][发送编号：[1575627633] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_01, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 49, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo01Message{id=1575627633}, timestamp=null), recordMetadata=DEMO_01-0@2]]]

# Demo01AConsumer 消费了一次该消息
2019-12-08 18:20:34.139  INFO 89818 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo01AConsumer       : [onMessage][线程编号:16 消息内容：ConsumerRecord(topic = DEMO_01, partition = 0, leaderEpoch = 0, offset = 2, CreateTime = 1575627634079, serialized key size = -1, serialized value size = 17, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = Demo01Message{id=1575627633})

# Demo01Consumer 消费了一次该消息
2019-12-08 18:20:34.142  INFO 89818 --- [ntainer#1-0-C-1] c.i.s.l.k.consumer.Demo01Consumer        : [onMessage][线程编号:18 消息内容：Demo01Message{id=1575627633}]

和 #testSyncSend() 方法执行的结果，是一致的。此时，我们打开 #testSyncSend() 方法所在的控制台，不会看到有消息消费的日志。说明，符合集群消费的机制：集群消费模式下，相同 Consumer Group 的每个 Consumer 实例平均分摊消息。。
😈 不过如上的日志，也可能出现在 #testSyncSend() 方法所在的控制台，而不在 #testASyncSend() 方法所在的控制台。

3.9 @KafkaListener

在「3.6 Demo01Consumer」中，我们已经使用了 @KafkaListener 注解，设置每个 Kafka 消费者 Consumer 的消息监听器的配置。

@KafkaListener 注解的常用属性如下：

/**
 * 监听的 Topic 数组
 * 
 * The topics for this listener.
 * The entries can be 'topic name', 'property-placeholder keys' or 'expressions'.
 * An expression must be resolved to the topic name.
 * This uses group management and Kafka will assign partitions to group members.
 * <p>
 * Mutually exclusive with {@link #topicPattern()} and {@link #topicPartitions()}.
 * @return the topic names or expressions (SpEL) to listen to.
 */
String[] topics() default {};
/**
 * 监听的 Topic 表达式
 * 
 * The topic pattern for this listener. The entries can be 'topic pattern', a
 * 'property-placeholder key' or an 'expression'. The framework will create a
 * container that subscribes to all topics matching the specified pattern to get
 * dynamically assigned partitions. The pattern matching will be performed
 * periodically against topics existing at the time of check. An expression must
 * be resolved to the topic pattern (String or Pattern result types are supported).
 * This uses group management and Kafka will assign partitions to group members.
 * <p>
 * Mutually exclusive with {@link #topics()} and {@link #topicPartitions()}.
 * @return the topic pattern or expression (SpEL).
 * @see org.apache.kafka.clients.CommonClientConfigs#METADATA_MAX_AGE_CONFIG
 */
String topicPattern() default "";
/**
 * @TopicPartition 注解的数组。每个 @TopicPartition 注解，可配置监听的 Topic、队列、消费的开始位置
 * 
 * The topicPartitions for this listener when using manual topic/partition
 * assignment.
 * <p>
 * Mutually exclusive with {@link #topicPattern()} and {@link #topics()}.
 * @return the topic names or expressions (SpEL) to listen to.
 */
TopicPartition[] topicPartitions() default {};

/**
 * 消费者分组
 * Override the {@code group.id} property for the consumer factory with this value
 * for this listener only.
 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
 * @return the group id.
 * @since 1.3
 */
String groupId() default "";

/**
 * 使用消费异常处理器 KafkaListenerErrorHandler 的 Bean 名字
 * 
 * Set an {@link org.springframework.kafka.listener.KafkaListenerErrorHandler} bean
 * name to invoke if the listener method throws an exception.
 * @return the error handler.
 * @since 1.3
 */
String errorHandler() default "";

/**
 * 自定义消费者监听器的并发数，这个我们在 TODO 详细解析。
 * 
 * Override the container factory's {@code concurrency} setting for this listener. May
 * be a property placeholder or SpEL expression that evaluates to a {@link Number}, in
 * which case {@link Number#intValue()} is used to obtain the value.
 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
 * @return the concurrency.
 * @since 2.2
 */
String concurrency() default "";

/**
 * 是否自动启动监听器。默认情况下，为 true 自动启动。
 *  
 * Set to true or false, to override the default setting in the container factory. May
 * be a property placeholder or SpEL expression that evaluates to a {@link Boolean} or
 * a {@link String}, in which case the {@link Boolean#parseBoolean(String)} is used to
 * obtain the value.
 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
 * @return true to auto start, false to not auto start.
 * @since 2.2
 */
String autoStartup() default "";

/**
 * Kafka Consumer 拓展属性。
 * 
 * Kafka consumer properties; they will supersede any properties with the same name
 * defined in the consumer factory (if the consumer factory supports property overrides).
 * <h3>Supported Syntax</h3>
 * <p>The supported syntax for key-value pairs is the same as the
 * syntax defined for entries in a Java
 * {@linkplain java.util.Properties#load(java.io.Reader) properties file}:
 * <ul>
 * <li>{@code key=value}</li>
 * <li>{@code key:value}</li>
 * <li>{@code key value}</li>
 * </ul>
 * {@code group.id} and {@code client.id} are ignored.
 * @return the properties.
 * @since 2.2.4
 * @see org.apache.kafka.clients.consumer.ConsumerConfig
 * @see #groupId()
 * @see #clientIdPrefix()
 */
String[] properties() default {};

@KafkaListener 注解的不常用属性如下：

/**
 * 唯一标识
 *  
 * The unique identifier of the container managing for this endpoint.
 * <p>If none is specified an auto-generated one is provided.
 * <p>Note: When provided, this value will override the group id property
 * in the consumer factory configuration, unless {@link #idIsGroup()}
 * is set to false.
 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
 * @return the {@code id} for the container managing for this endpoint.
 * @see org.springframework.kafka.config.KafkaListenerEndpointRegistry#getListenerContainer(String)
 */
String id() default "";
/**
 * id 唯一标识的前缀
 *  
 * When provided, overrides the client id property in the consumer factory
 * configuration. A suffix ('-n') is added for each container instance to ensure
 * uniqueness when concurrency is used.
 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
 * @return the client id prefix.
 * @since 2.1.1
 */
String clientIdPrefix() default "";
/**
 * 当 groupId 未设置时，是否使用 id 作为 groupId
 * 
 * When {@link #groupId() groupId} is not provided, use the {@link #id() id} (if
 * provided) as the {@code group.id} property for the consumer. Set to false, to use
 * the {@code group.id} from the consumer factory.
 * @return false to disable.
 * @since 1.3
 */
boolean idIsGroup() default true;

/**
 * 使用的 KafkaListenerContainerFactory Bean 的名字。
 * 若未设置，则使用默认的 KafkaListenerContainerFactory Bean 。
 * 
 * The bean name of the {@link org.springframework.kafka.config.KafkaListenerContainerFactory}
 * to use to create the message listener container responsible to serve this endpoint.
 * <p>If not specified, the default container factory is used, if any.
 * @return the container factory bean name.
 */
String containerFactory() default "";

/**
 * 所属 MessageListenerContainer Bean 的名字。
 * 
 * If provided, the listener container for this listener will be added to a bean
 * with this value as its name, of type {@code Collection<MessageListenerContainer>}.
 * This allows, for example, iteration over the collection to start/stop a subset
 * of containers.
 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
 * @return the bean name for the group.
 */
String containerGroup() default "";

/**
 * 真实监听容器的 Bean 名字，需要在名字前加 "__" 。
 * 
 * A pseudo bean name used in SpEL expressions within this annotation to reference
 * the current bean within which this listener is defined. This allows access to
 * properties and methods within the enclosing bean.
 * Default '__listener'.
 * <p>
 * Example: {@code topics = "#{__listener.topicList}"}.
 * @return the pseudo bean name.
 * @since 2.1.2
 */
String beanRef() default "__listener";

@TopicPartition 注解
@PartitionOffset 注解
@KafkaListeners 注解，允许我们在其中，同时添加多个 @KafkaListener 注解。

4. 批量发送消息

示例代码对应仓库：lab-03-kafka-demo-batch 。

在一些业务场景下，我们希望使用 Producer 批量发送消息，提高发送性能。不同于我们在《芋道 Spring Boot 消息队列 RocketMQ 入门》的「4. 批量发送消息」功能，RocketMQ 是提供了一个可以批量发送多条消息的 API 。而 Kafka 提供的批量发送消息，它提供了一个 RecordAccumulator 消息收集器，将发送给相同 Topic 的相同 Partition 分区的消息们，“偷偷”收集在一起，当满足条件时候，一次性批量发送提交给 Kafka Broker 。如下是三个条件，满足任一即会批量发送：

【数量】batch-size ：超过收集的消息数量的最大条数。
【空间】buffer-memory ：超过收集的消息占用的最大内存。
【时间】linger.ms ：超过收集的时间的最大等待时长，单位：毫秒。

下面，我们来实现一个 Producer 批量发送消息的示例。考虑到不污染「3. 快速入门」的示例，我们新建一个 lab-03-kafka-demo-batch 项目。

4.1 引入依赖

和 3.1 引入依赖」一致，见 pom.xml 文件。

4.2 应用配置文件

在 resources 目录下，创建 application.yaml 配置文件。配置如下：

spring:
  # Kafka 配置项，对应 KafkaProperties 配置类
  kafka:
    bootstrap-servers: 127.0.0.1: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 的序列化
      batch-size: 16384 # 每次批量发送消息的最大数量
      buffer-memory: 33554432 # 每次批量发送消息的最大内存
      properties:
        linger:
          ms: 30000 # 批处理延迟时间上限。这里配置为 30 * 1000 ms 过后，不管是否消息数量是否到达 batch-size 或者消息大小到达 buffer-memory 后，都直接发送一次请求。
    # Kafka Consumer 配置项
    consumer:
      auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring:
          json:
            trusted:
              packages: cn.iocoder.springboot.lab03.kafkademo.message
    # Kafka Consumer Listener 监听器配置
    listener:
      missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错

logging:
  level:
    org:
      springframework:
        kafka: ERROR # spring-kafka INFO 日志太多了，所以我们限制只打印 ERROR 级别
      apache:
        kafka: ERROR # kafka INFO 日志太多了，所以我们限制只打印 ERROR 级别

相比「3.2 应用配置文件」来说，额外三个参数，就是我们说的 Producer 批量发送的三个条件：
- spring.kafka.producer.batch-size
- spring.kafka.producer.buffer-memory
- spring.kafka.producer.properties.linger.ms
具体的数值配置多少，根据自己的应用来。这里，我们故意将 linger.ms 配置成了 30 秒，主要为了演示之用。

4.3 Demo02Message

在 cn.iocoder.springboot.lab03.kafkademo.message 包下，创建 Demo02Message 消息类，提供给当前示例使用。代码如下：

// Demo02Message.java

public class Demo02Message {

    public static final String TOPIC = "DEMO_012";

    /**
     * 编号
     */
    private Integer id;

    // ... 省略 set/get/toString 方法

}

TOPIC 静态属性，我们设置该消息类对应 Topic 为 "DEMO_02" 。
其它都和「3.4 Demo01Message」是一样的。

4.4 Demo02Producer

在 cn.iocoder.springboot.lab03.kafkademo.producer 包下，创建 Demo02Producer 类，它会使用 Kafka-Spring 封装提供的 KafkaTemplate ，实现一个异步发送消息的方法。代码如下：

// Demo02Producer.java

@Component
public class Demo02Producer {

    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;

    public ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id) {
        // 创建 Demo02Message 消息
        Demo02Message message = new Demo02Message();
        message.setId(id);
        // 异步发送消息
        return kafkaTemplate.send(Demo02Message.TOPIC, message);
    }

}

看起来和我们在「3.5 Demo01Producer」提供的异步发送消息的方法，除了换成了 Demo02Message 消息对象，其它都是一模一样的。😈 对的，这也是为什么艿艿在上文说到，Kafka 是“偷偷”收集来实现批量发送，对于我们使用发送消息的方法，还是一致的。
因为我们发送的消息 Topic 是自动创建的，所以其 Partition 分区大小是 1 。这样，就能保证我每次调用这个方法，满足批量发送消息的一个前提，相同 Topic 的相同 Partition 分区的消息们。

4.5 Demo02Consumer

在 cn.iocoder.springboot.lab03.kafkademo.consumer 包下，创建 Demo02Consumer 类，消费消息。代码如下：

// Demo02Consumer.java

@Component
public class Demo02Consumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = Demo02Message.TOPIC,
            groupId = "demo02-consumer-group-" + Demo02Message.TOPIC)
    public void onMessage(Demo02Message message) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

}

和「3.6 Demo01Consumer」基本一直，除了是不同的消费者分组，消费了不同的 Topic 。

4.6 简单测试

创建 Demo02ProducerTest 测试类，编写单元测试方法，测试 Producer 批量发送消息的效果。代码如下：

// Demo02ProducerTest.java

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class Demo02ProducerTest {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private Demo02Producer producer;

    @Test
    public void testASyncSend() throws InterruptedException {
        logger.info("[testASyncSend][开始执行]");

        for (int i = 0; i < 3; i++) {
            int id = (int) (System.currentTimeMillis() / 1000);
            producer.asyncSend(id).addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {

                @Override
                public void onFailure(Throwable e) {
                    logger.info("[testASyncSend][发送编号：[{}] 发送异常]]", id, e);
                }

                @Override
                public void onSuccess(SendResult<Object, Object> result) {
                    logger.info("[testASyncSend][发送编号：[{}] 发送成功，结果为：[{}]]", id, result);
                }

            });

            // 故意每条消息之间，隔离 10 秒
            Thread.sleep(10 * 1000L);
        }

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }

}

异步发送三条消息，每次发送消息之间，都故意 sleep 10 秒。😈 目的是，恰好满足我们配置的 linger.ms 最大等待时长。

我们来执行 #testASyncSend() 方法，测试批量发送消息。控制台输出如下：

# 打印 testASyncSend 方法开始执行的日志
2019-12-08 21:43:02.330  INFO 94957 --- [           main] c.i.s.l.k.producer.Demo02ProducerTest    : [testASyncSend][开始执行]

# 30 秒后，满足批量消息的最大等待时长，所以 3 条消息被 Producer 批量发送。
# 因此我们配置的是 acks=1 ，需要等待发送成功后，才会回调 ListenableFutureCallback 的方法。
2019-12-08 21:43:32.424  INFO 94957 --- [ad | producer-1] c.i.s.l.k.producer.Demo02ProducerTest    : [testASyncSend][发送编号：[1575639782] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_02, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 50, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo01Message{id=1575639782}, timestamp=null), recordMetadata=DEMO_02-0@37]]]
2019-12-08 21:43:32.425  INFO 94957 --- [ad | producer-1] c.i.s.l.k.producer.Demo02ProducerTest    : [testASyncSend][发送编号：[1575639792] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_02, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 50, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo01Message{id=1575639792}, timestamp=null), recordMetadata=DEMO_02-0@38]]]
2019-12-08 21:43:32.425  INFO 94957 --- [ad | producer-1] c.i.s.l.k.producer.Demo02ProducerTest    : [testASyncSend][发送编号：[1575639802] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_02, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 50, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo01Message{id=1575639802}, timestamp=null), recordMetadata=DEMO_02-0@39]]]

# 因为 Producer 批量发送完成，所以 Demo02Consumer 消费到消息
2019-12-08 21:43:32.475  INFO 94957 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo02Consumer        : [onMessage][线程编号:16 消息内容：Demo01Message{id=1575639782}]
2019-12-08 21:43:32.475  INFO 94957 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo02Consumer        : [onMessage][线程编号:16 消息内容：Demo01Message{id=1575639792}]
2019-12-08 21:43:32.475  INFO 94957 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo02Consumer        : [onMessage][线程编号:16 消息内容：Demo01Message{id=1575639802}]

😈 胖友认真看下艿艿在日志中的注释，理解下整个批量发送消息的过程。不过还是那句话，实际场景下，我们不太会把 linger.ms 配置的这么长时间，这里仅仅是演示。

5. 批量消费消息

示例代码对应仓库：lab-03-kafka-demo-batch-consume 。

在一些业务场景下，我们希望使用 Consumer 批量消费消息，提高消费速度。要注意，Consumer 的批量消费消息，和 Producer 的「4. 批量发送消息」没有直接关联哈。

下面，我们来实现一个 Consumer 批量消费消息的示例。考虑到不污染「4. 批量发送消息」的示例，我们在 lab-03-kafka-demo-batch 项目的基础上，复制出一个 lab-03-kafka-demo-batch-consume 项目。😈 酱紫，我们也能少写点代码，哈哈哈~

5.1 应用配置文件

修改 application.yaml 配置文件。配置如下：

spring:
  # Kafka 配置项，对应 KafkaProperties 配置类
  kafka:
    bootstrap-servers: 127.0.0.1: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 的序列化
      batch-size: 16384 # 每次批量发送消息的最大数量
      buffer-memory: 33554432 # 每次批量发送消息的最大内存
      properties:
        linger:
          ms: 30000 # 批处理延迟时间上限。这里配置为 30 * 1000 ms 过后，不管是否消息数量是否到达 batch-size 或者消息大小到达 buffer-memory 后，都直接发送一次请求。
    # Kafka Consumer 配置项
    consumer:
      auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      fetch-max-wait: 10000 # poll 一次拉取的阻塞的最大时长，单位：毫秒。这里指的是阻塞拉取需要满足至少 fetch-min-size 大小的消息
      fetch-min-size: 10 # poll 一次消息拉取的最小数据量，单位：字节
      max-poll-records: 100 # poll 一次消息拉取的最大数量
      properties:
        spring:
          json:
            trusted:
              packages: cn.iocoder.springboot.lab03.kafkademo.message
    # Kafka Consumer Listener 监听器配置
    listener:
      type: BATCH # 监听器类型，默认为 SINGLE ，只监听单条消息。这里我们配置 BATCH ，监听多条消息，批量消费
      missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错

logging:
  level:
    org:
      springframework:
        kafka: ERROR # spring-kafka INFO 日志太多了，所以我们限制只打印 ERROR 级别
      apache:
        kafka: ERROR # kafka INFO 日志太多了，所以我们限制只打印 ERROR 级别

相比「3.2 应用配置文件」来说，增加了四个配置项，胖友自己根据注释，自己理解下噢。
- spring.kafka.listener.type
- spring.kafka.consumer.max-poll-records
- spring.kafka.consumer.fetch-min-size
- spring.kafka.consumer.fetch-max-wait

5.2 Demo02Consumer

修改 Demo02Consumer 消费者，改成批量消费消息。代码如下：

// Demo02Consumer.java

@Component
public class Demo02Consumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = Demo02Message.TOPIC,
            groupId = "demo02-consumer-group-" + Demo02Message.TOPIC)
    public void onMessage(List<Demo02Message> messages) {
        logger.info("[onMessage][线程编号:{} 消息数量：{}]", Thread.currentThread().getId(), messages.size());
    }

}

相比「4.5 Demo02Consumer」来说，方法上的参数变成了 List 数组。

5.3 简单测试

还是使用 Demo02ProducerTest 测试类，执行单元测试，输出日志如下：

2019-12-08 23:00:14.274  INFO 98637 --- [           main] c.i.s.l.k.producer.Demo02ProducerTest    : [testASyncSend][开始执行]
2019-12-08 23:00:44.385  INFO 98637 --- [ad | producer-1] c.i.s.l.k.producer.Demo02ProducerTest    : [testASyncSend][发送编号：[1575644414] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_02, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 50, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo01Message{id=1575644414}, timestamp=null), recordMetadata=DEMO_02-0@55]]]
2019-12-08 23:00:44.386  INFO 98637 --- [ad | producer-1] c.i.s.l.k.producer.Demo02ProducerTest    : [testASyncSend][发送编号：[1575644424] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_02, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 50, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo01Message{id=1575644424}, timestamp=null), recordMetadata=DEMO_02-0@56]]]
2019-12-08 23:00:44.387  INFO 98637 --- [ad | producer-1] c.i.s.l.k.producer.Demo02ProducerTest    : [testASyncSend][发送编号：[1575644434] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_02, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 50, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo01Message{id=1575644434}, timestamp=null), recordMetadata=DEMO_02-0@57]]]

# 批量消费了 3 条消息
2019-12-08 23:00:44.425  INFO 98637 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo02Consumer        : [onMessage][线程编号:16 消息数量：3]

从日志中，我们可以看出，发送的 3 条消息被 Demo02Consumer 批量消费了。
😈 为了更好的做对比，胖友可以尝试自行把配置改成 spring.kafka.listener.type=SINGLE ，就会发现 Demo02Consumer 只会单条消费了。

6. 定时消息

Kafka 并未提供定时消息的功能，需要我们自行拓展。

例如说《基于 Kafka 的定时消息/任务服》文章，提供的方案。

当然，也可以考虑基于 MySQL 存储定时消息，Job 扫描到达时间的定时消息，发送给 Kafka 。

7. 消费重试

示例代码对应仓库：lab-31-kafka-demo 。

Spring-Kafka 提供消费重试的机制。在消息消费失败的时候，Spring-Kafka 会通过消费重试机制，重新投递该消息给 Consumer ，让 Consumer 有机会重新消费消息，实现消费成功。

当然，Spring-Kafka 并不会无限重新投递消息给 Consumer 重新消费，而是在默认情况下，达到 N 次重试次数时，Consumer 还是消费失败时，该消息就会进入到死信队列。

死信队列用于处理无法被正常消费的消息。当一条消息初次消费失败，Spring-Kafka 会自动进行消息重试；达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，Spring-Kafka 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。

Spring-Kafka 将这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。后续，我们可以通过对死信队列中的消息进行重发，来使得消费者实例再次进行消费。

在《芋道 Spring Boot 消息队列 RocketMQ 入门》的「6. 消费重试」小节中，我们可以看到，消费重试和死信队列，是 RocketMQ 自带的功能。
而在 Kafka 中，消费重试和死信队列，是由 Spring-Kafka 所封装提供的。

每条消息的失败重试，是可以配置一定的间隔时间。具体，我们在示例的代码中，来进行具体的解释。

下面，我们开始本小节的示例。该示例，我们会在「3. 快速入门」的 lab-31-kafka-demo 项目中，继续改造。

7.1 KafkaConfiguration

在 cn.iocoder.springboot.lab03.kafkademo.config 包下，创建 KafkaConfiguration 配置类，增加消费异常的 ErrorHandler 处理器。代码如下：

// KafkaConfiguration.java

@Configuration
public class KafkaConfiguration {

    @Bean
    @Primary
    public ErrorHandler kafkaErrorHandler(KafkaTemplate<?, ?> template) {
        // <1> 创建 DeadLetterPublishingRecoverer 对象
        ConsumerRecordRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
        // <2> 创建 FixedBackOff 对象
        BackOff backOff = new FixedBackOff(10 * 1000L, 3L);
        // <3> 创建 SeekToCurrentErrorHandler 对象
        return new SeekToCurrentErrorHandler(recoverer, backOff);
    }

}

Spring-Kafka 的消费重试功能，通过实现自定义的 SeekToCurrentErrorHandler ，在 Consumer 消费消息异常的时候，进行拦截处理：
- 在重试小于最大次数时，重新投递该消息给 Consumer ，让 Consumer 有机会重新消费消息，实现消费成功。
- 在重试到达最大次数时，Consumer 还是消费失败时，该消息就会发送到死信队列。例如说，本小节我们测试的 Topic 是 "DEMO_04" ，则其对应的死信队列的 Topic 就是 "DEMO_04.DLT" ，即在原有 Topic 加上 .DLT 后缀，就是其死信队列的 Topic 。
<1> 处，创建 DeadLetterPublishingRecoverer 对象，它负责实现，在重试到达最大次数时，Consumer 还是消费失败时，该消息就会发送到死信队列。
<2> 处，创建 FixedBackOff 对象。这里，我们配置了重试 3 次，每次固定间隔 30 秒。当然，胖友可以选择 BackOff 的另一个子类 ExponentialBackOff 实现，提供指数递增的间隔时间。
<3> 处，创建 SeekToCurrentErrorHandler 对象，负责处理异常，串联整个消费重试的整个过程。

这里，我们来简单说说 SeekToCurrentErrorHandler 是怎么提供消费重试的功能的。

在消息消费失败时，SeekToCurrentErrorHandler 会将调用 Kafka Consumer 的 #seek(TopicPartition partition, long offset) 方法，将 Consumer 对于该消息对应的 TopicPartition 分区的本地进度设置成该消息的位置。这样，Consumer 在下次从 Kafka Broker 拉取消息的时候，又能重新拉取到这条消费失败的消息，并且是第一条。
同时，Spring-Kafka 使用 FailedRecordTracker 对每个 Topic 的每个 TopicPartition 消费失败次数进行计数，这样相当于对该 TopicPartition 的第一条消费失败的消息的消费失败次数进行计数。😈 这里，胖友好好思考下，结合艿艿在上一点的描述。
另外，在 FailedRecordTracker 中，会调用 BackOff 来进行计算，该消息的下一次重新消费的时间，通过 Thread#sleep(...) 方法，实现重新消费的时间间隔。
有一点需要注意，FailedRecordTracker 提供的计数是客户端级别的，重启 JVM 应用后，计数是会丢失的。所以，如果想要计数进行持久化，需要自己重新实现下 FailedRecordTracker 类，通过 ZooKeeper 存储计数。

😈 RocketMQ 提供的消费重试的计数，目前是服务端级别，已经进行持久化。

对了，SeekToCurrentErrorHandler 是只针对消息的单条消费失败的消费重试处理。如果胖友想要有消息的批量消费失败的消费重试处理，可以使用 SeekToCurrentBatchErrorHandler 。配置方式如下：

@Bean
@Primary
public BatchErrorHandler kafkaBatchErrorHandler() {
    // 创建 SeekToCurrentBatchErrorHandler 对象
    SeekToCurrentBatchErrorHandler batchErrorHandler = new SeekToCurrentBatchErrorHandler();
    // 创建 FixedBackOff 对象
    BackOff backOff = new FixedBackOff(10 * 1000L, 3L);
    batchErrorHandler.setBackOff(backOff);
    // 返回
    return batchErrorHandler;
}

从代码中我们可以看到，并没有设置 DeadLetterPublishingRecoverer 对象。因为 SeekToCurrentBatchErrorHandler 暂时不支持死信队列的机制。

另外，如果胖友想要自定义 ErrorHandler 或 BatchErrorHandler 实现类，实现对消费异常的自定义的逻辑，也是可以的。

实现的代码的示例，可以参考 LoggingErrorHandler 类。
配置的方式，和本小节配置 SeekToCurrentErrorHandler 或 SeekToCurrentBatchErrorHandler 是一样的。
当然，绝大多数情况下，我们使用 SeekToCurrentErrorHandler 或 SeekToCurrentBatchErrorHandler 是足够的。

艿艿：貌似本小节信息量，略微有一点点大，胖友可以自己好好消化下。同时，也可以调试下整个过程涉及到的源码，更加具象下。「源码之前，了无秘密」。

7.2 Demo04Message

在 cn.iocoder.springboot.lab03.kafkademo.message 包下，创建 Demo04Message 消息类，提供给当前示例使用。代码如下：

// Demo04Message.java

public class Demo04Message {

    public static final String TOPIC = "DEMO_04";

    /**
     * 编号
     */
    private Integer id;

    // ... 省略 set/get/toString 方法

}

TOPIC 静态属性，我们设置该消息类对应 Topic 为 "DEMO_04" 。

7.3 Demo04Producer

在 cn.iocoder.springboot.lab03.kafkademo.producer 包下，创建 Demo04Producer 类，它会使用 Kafka-Spring 封装提供的 KafkaTemplate ，同步发送消息。代码如下：

// Demo04Producer.java

@Component
public class Demo04Producer {

    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;

    public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
        // 创建 Demo04Message 消息
        Demo04Message message = new Demo04Message();
        message.setId(id);
        // 同步发送消息
        return kafkaTemplate.send(Demo04Message.TOPIC, message).get();
    }

}

和「3.5 Demo01Producer」的同步发送消息的代码是一致的，就是换成了 Demo04Message 。

7.4 Demo04Consumer

在 cn.iocoder.springboot.lab03.kafkademo.consumer 包下，创建 Demo04Consumer 类，消费消息。代码如下：

// Demo04Consumer.java

@Component
public class Demo04Consumer {

    private AtomicInteger count = new AtomicInteger(0);

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = Demo04Message.TOPIC,
            groupId = "demo04-consumer-group-" + Demo04Message.TOPIC)
    public void onMessage(Demo04Message message) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
        // <X> 注意，此处抛出一个 RuntimeException 异常，模拟消费失败
        throw new RuntimeException("我就是故意抛出一个异常");
    }

}

在 <X> 处，我们在消费消息时候，抛出一个 RuntimeException 异常，模拟消费失败。

7.5 简单测试

创建 Demo04ProducerTest 测试类，编写一个单元测试方法，调用 Demo04Producer 同步发送消息。代码如下：

// Demo04ProducerTest.java

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class Demo04ProducerTest {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private Demo04Producer producer;

    @Test
    public void testSyncSend() throws ExecutionException, InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        SendResult result = producer.syncSend(id);
        logger.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }

}

我们来执行 #testSyncSend() 方法，同步发送消息。控制台输出如下：

# Producer 同步发送消息成功
2019-12-07 10:24:18.851  INFO 11359 --- [           main] c.i.s.l.k.producer.Demo04ProducerTest    : [testSyncSend][发送编号：[1575685458] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=DEMO_04, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 52, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo04Message{id=1575685458}, timestamp=null), recordMetadata=DEMO_04-0@0]]]

# Consumer04 首次消费
2019-12-07 10:24:18.918  INFO 11359 --- [ntainer#2-0-C-1] c.i.s.l.k.consumer.Demo04Consumer        : [onMessage][线程编号:16 消息内容：Demo04Message{id=1575685458}]
# 打印异常
2019-12-07 10:24:28.927 ERROR 11359 --- [ntainer#2-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception

# Consumer04 第一次重试消费
2019-12-07 10:24:28.929  INFO 11359 --- [ntainer#2-0-C-1] c.i.s.l.k.consumer.Demo04Consumer        : [onMessage][线程编号:16 消息内容：Demo04Message{id=1575685458}]
# 打印异常
2019-12-07 10:24:38.932 ERROR 11359 --- [ntainer#2-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception

# Consumer04 第二次重试消费
2019-12-07 10:24:38.934  INFO 11359 --- [ntainer#2-0-C-1] c.i.s.l.k.consumer.Demo04Consumer        : [onMessage][线程编号:16 消息内容：Demo04Message{id=1575685458}]
# 打印异常
2019-12-07 10:24:48.939 ERROR 11359 --- [ntainer#2-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception

# Consumer04 第三次重试消费
2019-12-07 10:24:48.941  INFO 11359 --- [ntainer#2-0-C-1] c.i.s.l.k.consumer.Demo04Consumer        : [onMessage][线程编号:16 消息内容：Demo04Message{id=1575685458}]
# 这次不会打印异常日志，直接发到死信队列

从日志中，我们可以看出，发送的这条消息被 Demo04Consumer 首次消费失败后，每间隔 10 秒，又消费重试 3 次。

8. 广播消费

示例代码对应仓库：lab-03-kafka-demo-broadcast 。

在上述的示例中，我们看到的都是使用集群消费。而在一些场景下，我们需要使用广播消费。

广播消费模式下，相同 Consumer Group 的每个 Consumer 实例都接收全量的消息。

不过 Kafka 并不直接提供内置的广播消费的功能！！！此时，我们只能退而求其次，每个 Consumer 独有一个 Consumer Group ，从而保证都能接收到全量的消息。

例如说，在应用中，缓存了数据字典等配置表在内存中，可以通过 Kafka 广播消费，实现每个应用节点都消费消息，刷新本地内存的缓存。

又例如说，我们基于 WebSocket 实现了 IM 聊天，在我们给用户主动发送消息时，因为我们不知道用户连接的是哪个提供 WebSocket 的应用，所以可以通过 Kafka 广播消费，每个应用判断当前用户是否是和自己提供的 WebSocket 服务连接，如果是，则推送消息给用户。

下面，我们开始本小节的示例。考虑到不污染上述的示例，我们新建一个 lab-03-kafka-demo-broadcast 项目。

8.1 引入依赖

和「3.1 引入依赖」」一致，见 pom.xml 文件。

8.2 应用配置文件

在「3.2 应用配置文件」是一致的订单，就是修改了配置项 spring.kafka.consumer.auto-offset-reset=latest 。因为在广播订阅下，我们一般情况下，无需消费历史的消息，而是从订阅的 Topic 的队列的尾部开始消费即可，所以配置为 latest 。

完整的配置文件，见 application.yaml 。

8.3 Demo05Message

在 cn.iocoder.springboot.lab03.kafkademo.message 包下，创建 Demo05Message 消息类，提供给当前示例使用。代码如下：

// Demo05Message.java

public class Demo05Message {

    public static final String TOPIC = "DEMO_05";

    /**
     * 编号
     */
    private Integer id;

    // ... 省略 set/get/toString 方法

}

TOPIC 静态属性，我们设置该消息类对应 Topic 为 "DEMO_05" 。

8.4 Demo05Producer

在 cn.iocoder.springboot.lab03.kafkademo.producer 包下，创建 Demo05Producer 类，它会使用 Kafka-Spring 封装提供的 KafkaTemplate ，同步发送消息。代码如下：

// Demo04Producer.java

@Component
public class Demo05Producer {

    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;

    public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
        // 创建 Demo05Message 消息
        Demo05Message message = new Demo05Message();
        message.setId(id);
        // 同步发送消息
        return kafkaTemplate.send(Demo05Message.TOPIC, message).get();
    }

}

和「3.5 Demo01Producer」的同步发送消息的代码是一致的，就是换成了 Demo05Message 。

8.5 Demo05Consumer

在 cn.iocoder.springboot.lab03.kafkademo.consumer 包下，创建 Demo05Consumer 类，消费消息。代码如下：

// Demo04Consumer.java

@Component
public class Demo05Consumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = Demo05Message.TOPIC,
            groupId = "demo05-consumer-group-" + Demo05Message.TOPIC + "-" + "#{T(java.util.UUID).randomUUID()}") // <X>
    public void onMessage(Demo05Message message) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

}

在 <X> 处，我们通过 Spring EL 表达式，在每个消费者分组的名字上，使用 UUID 生成其后缀。这样，我们就能保证每个项目启动的消费者分组不同，以达到广播消费的目的。

8.6 简单测试

创建 Demo05ProducerTest 测试类，用于测试广播消费。代码如下：

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class Demo05ProducerTest {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private Demo05Producer producer;

    @Test
    public void test() throws InterruptedException {
        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }

    @Test
    public void testSyncSend() throws ExecutionException, InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        SendResult result = producer.syncSend(id);
        logger.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }

}

首先，执行 #test() 测试方法，先启动一个消费者分组 "demo05-consumer-group-DEMO_05-${UUID1}" 的 Consumer 节点。

然后，执行 #testSyncSend() 测试方法，再启动一个消费者分组 "demo05-consumer-group-DEMO_05-${UUID2}" 的 Consumer 节点。同时，该测试方法，调用 Demo05ProducerTest#syncSend(id) 方法，同步发送了一条消息。控制台输出如下：

// #### testSyncSend 方法对应的控制台 ####

# Producer 同步发送消息成功
2019-12-07 15:00:42.578  INFO 16077 --- [           main] c.i.s.l.k.producer.Demo05ProducerTest    : [testSyncSend][发送编号：[1575702042] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=DEMO_05, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 53, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo05Message{id=1575702042}, timestamp=null), recordMetadata=DEMO_05-0@0]]]

# Demo05Consumer 消费了该消息
2019-12-07 15:00:42.618  INFO 16077 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo05Consumer        : [onMessage][线程编号:16 消息内容：Demo05Message{id=1575702042}]

// ### test 方法对应的控制台 ####

# Demo05Consumer 也消费了该消息
2019-12-07 15:00:42.644  INFO 16067 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo05Consumer        : [onMessage][线程编号:16 消息内容：Demo05Message{id=1575702042}]

消费者分组 "demo05-consumer-group-DEMO_05-${UUID1}" 和 demo05-consumer-group-DEMO_05-${UUID2} 的两个 Consumer 节点，都消费了这条发送的消息。符合广播消费的预期~

9. 并发消费

示例代码对应仓库：lab-03-kafka-demo-concurrency 。

在上述的示例中，我们配置的每一个 Spring-Kafka @KafkaListener ，都是串行消费的。显然，这在监听的 Topic 每秒消息量比较大的时候，会导致消费不及时，导致消息积压的问题。

虽然说，我们可以通过启动多个 JVM 进程，实现多进程的并发消费，从而加速消费的速度。但是问题是，否能够实现多线程的并发消费呢？答案是有。

在「3.9 @KafkaListener」小节中，我们可以看到该注解有 concurrency 属性，它可以指定并发消费的线程数。例如说，如果设置 concurrency=4 时，Spring-Kafka 就会为该 @KafkaListener 创建 4 个线程，进行并发消费。

考虑到让胖友能够更好的理解 concurrency 属性，我们来简单说说 Spring-Kafka 在这块的实现方式。我们来举个例子：

首先，我们来创建一个 Topic 为 "DEMO_06" ，并且设置其 Partition 分区数为 10 。
然后，我们创建一个 Demo06Consumer 类，并在其消费方法上，添加 @KafkaListener(concurrency=2) 注解。
再然后，我们启动项目。Spring-Kafka 会根据 @KafkaListener(concurrency=2) 注解，创建 2 个 Kafka Consumer 。注意噢，是 2 个 Kafka Consumer 呢！！！后续，每个 Kafka Consumer 会被单独分配到一个线程中，进行拉取消息，消费消息。
之后，Kafka Broker 会将 Topic 为 "DEMO_06" 分配给创建的 2 个 Kafka Consumer 各 5 个 Partition 。😈 如果不了解 Kafka Broker “分配区分”机制单独胖友，可以看看《Kafka 消费者如何分配分区》文章。
这样，因为 @KafkaListener(concurrency=2) 注解，创建 2 个 Kafka Consumer ，就在各自的线程中，拉取各自的 Topic 为 "DEMO_06" 的 Partition 的消息，各自串行消费。从而，实现多线程的并发消费。

酱紫讲解一下，胖友对 Spring-Kafka 实现多线程的并发消费的机制，是否理解了。不过有一点要注意，不要配置 concurrency 属性过大，则创建的 Kafka Consumer 分配不到消费 Topic 的 Partition 分区，导致不断的空轮询。

友情提示：可以选择不看。

在理解 Spring-Kafka 提供的并发消费机制，花费了好几个小时，主要陷入到了一个误区。

如果胖友有使用过 RocketMQ 的并发消费，会发现只要创建一个 RocketMQ Consumer 对象，然后 Consumer 拉取完消息之后，丢到 Consumer 的线程池中执行消费，从而实现并发消费。

而在 Spring-Kafka 提供的并发消费，会发现需要创建多个 Kafka Consumer 对象，并且每个 Consumer 都单独分配一个线程，然后 Consumer 拉取完消息之后，在各自的线程中执行消费。

又或者说，Spring-Kafka 提供的并发消费，很像 RocketMQ 的顺序消费。😈 从感受上来说，Spring-Kafka 的并发消费像 BIO ，RocketMQ 的并发消费像 NIO 。

不过，理论来说，在原生的 Kafka 客户端，也是能封装出和 RocketMQ Consumer 一样的并发消费的机制。

也因此，在使用 Kafka 的时候，每个 Topic 的 Partition 在消息量大的时候，要注意设置的相对大一些。

下面，我们开始本小节的示例。本示例就是上述举例的具体实现。考虑到不污染上述的示例，我们新建一个 lab-03-kafka-demo-concurrency 项目。

9.1 引入依赖

和 3.1 引入依赖」一致，见 pom.xml 文件。

9.2 应用配置文件

和 3.2 应用配置文件」一致，见 application.yaml 文件。

实际上，可以通过 spring.kafka.listener.concurrency 配置项，全局设置每个 @KafkaListener 的并发消费的线程数。不过个人建议，还是每个 @KafkaListener 各自配置，毕竟每个 Topic 的 Partition 的数量，都是不同的。当然，也可以结合使用 😈 。

9.3 Demo06Message

在 cn.iocoder.springboot.lab03.kafkademo.message 包下，创建 Demo06Message 消息类，提供给当前示例使用。代码如下：

// Demo06Message.java

public class Demo06Message {

    public static final String TOPIC = "DEMO_06";

    /**
     * 编号
     */
    private Integer id;

    // ... 省略 set/get/toString 方法

}

TOPIC 静态属性，我们设置该消息类对应 Topic 为 "DEMO_06" 。

注意，记得手动创建一个 "DEMO_06" 的 Partition 大小为 10 。可执行如下命令：

$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 10 --topic DEMO_06

9.4 Demo06Producer

在 cn.iocoder.springboot.lab03.kafkademo.producer 包下，创建 Demo06Producer 类，它会使用 Kafka-Spring 封装提供的 KafkaTemplate ，同步发送消息。代码如下：

// Demo06Producer.java

@Component
public class Demo06Producer {

    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;

    public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
        // 创建 Demo01Message 消息
        Demo06Message message = new Demo06Message();
        message.setId(id);
        // 同步发送消息
        return kafkaTemplate.send(Demo06Message.TOPIC, message).get();
    }

}

和「3.5 Demo01Producer」的同步发送消息的代码是一致的，就是换成了 Demo06Message 。

9.5 Demo06Consumer

在 cn.iocoder.springboot.lab03.kafkademo.consumer 包下，创建 Demo06Consumer 类，消费消息。代码如下：

// Demo06Consumer.java

@Component
public class Demo06Consumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = Demo06Message.TOPIC,
            groupId = "demo06-consumer-group-" + Demo06Message.TOPIC,
            concurrency = "2")
    public void onMessage(Demo06Message message) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

}

在 <X> 处，我们在 @KafkaListener 注解上，添加了 concurrency = "2" 属性，创建 2 个线程消费 "DEMO_06" 下的消息。

9.6 简单测试

创建 Demo06ProducerTest 测试类，编写一个单元测试方法，发送 10 条消息，观察并发消费情况。代码如下：

// Demo06ProducerTest.java

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class Demo06ProducerTest {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private Demo06Producer producer;

    @Test
    public void testSyncSend() throws ExecutionException, InterruptedException {
        for (int i = 0; i < 10; i++) {
            int id = (int) (System.currentTimeMillis() / 1000);
            SendResult result = producer.syncSend(id);
//        logger.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);
        }

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }

}

执行 #testSyncSend() 单元测试，输出日志如下：

# 线程编号为 16 
2019-12-07 17:21:16.365  INFO 24303 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:16 消息内容：Demo06Message{id=1575710476}]
2019-12-07 17:21:16.369  INFO 24303 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:16 消息内容：Demo06Message{id=1575710476}]
2019-12-07 17:21:16.369  INFO 24303 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:16 消息内容：Demo06Message{id=1575710476}]
2019-12-07 17:21:16.369  INFO 24303 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:16 消息内容：Demo06Message{id=1575710476}]
2019-12-07 17:21:16.369  INFO 24303 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:16 消息内容：Demo06Message{id=1575710476}]

# 线程编号为 18
2019-12-07 17:21:16.365  INFO 24303 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1575710476}]
2019-12-07 17:21:16.369  INFO 24303 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1575710476}]
2019-12-07 17:21:16.369  INFO 24303 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1575710476}]
2019-12-07 17:21:16.369  INFO 24303 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1575710476}]
2019-12-07 17:21:16.369  INFO 24303 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1575710476}]

我们可以看到，两个线程在消费 "DEMO_06" 下的消息。
😈 为了更好的做对比，胖友可以尝试自行把配置改成 @KafkaListner 的 concurrency=1，就会发现 Demo06Consumer 只会单线程消费了。

此时，如果我们使用 Kafka Manager 来查看 "DEMO_06" 的消费者列表：

10. 顺序消息

示例代码对应仓库：lab-03-kafka-demo-concurrency 。

我们先来一起了解下顺序消息的顺序消息的定义：

普通顺序消息：Producer 将相关联的消息发送到相同的消息队列。
完全严格顺序：在【普通顺序消息】的基础上，Consumer 严格顺序消费。

在上述的示例中，我们看到 Spring-Kafka 在 Consumer 消费消息时，天然就支持按照 Topic 下的 Partition 下的消息，顺序消费。即使在「9. 并发消费」时，也能保证如此。

那么此时，我们只需要考虑将 Producer 将相关联的消息发送到 Topic 下的相同的 Partition 即可。如果胖友了解 Producer 发送消息的分区策略的话，只要我们发送消息时，指定了消息的 key ，Producer 则会根据 key 的哈希值取模来获取到其在 Topic 下对应的 Partition 。完美~~不了解的 Producer 分区选择策略的胖友，可以看看《Kafka 发送消息分区选择策略详解》文章。

下面，我们开始本小节的示例。该示例，我们会在「9. 并发消费」的 lab-03-kafka-demo-concurrency 项目中，继续改造。

10.1 Demo06Producer

修改 Demo06Producer 类，增加顺序发送消息方法。代码如下：

// Demo06Producer.java

public SendResult syncSendOrderly(Integer id) throws ExecutionException, InterruptedException {
    // 创建 Demo01Message 消息
    Demo06Message message = new Demo06Message();
    message.setId(id);
    // 同步发送消息
    // 因为我们使用 String 的方式序列化 key ，所以需要将 id 转换成 String
    return kafkaTemplate.send(Demo06Message.TOPIC, String.valueOf(id), message).get();
}

调用 KafkaTemplate 同步发送消息方法时，我们多传入了 id 作为消息的 key ，从而实现发送到 DEMO_06 这个 Topic 下的相同 Partition 中。

10.2 简单测试

修改 Demo06ProducerTest 测试类，新增一个单元测试方法，顺序发送 10 条消息，观察消费情况。代码如下：

// Demo06ProducerTest.java

@Test
public void testSyncSendOrderly() throws ExecutionException, InterruptedException {
    for (int i = 0; i < 10; i++) {
        int id = 1;
        SendResult result = producer.syncSendOrderly(id);
        logger.info("[testSyncSend][发送编号：[{}] 发送队列：[{}]]", id, result.getRecordMetadata().partition());
    }

    // 阻塞等待，保证消费
    new CountDownLatch(1).await();
}

执行 #testSyncSendOrderly() 单元测试，输出日志如下：

# Producer 同步发送 10 条顺序消息成功，都发送到了 Topic 为 DEMO_06 ，队列编号为 9 的消息队列上
2019-12-07 18:48:45.866  INFO 31773 --- [           main] c.i.s.l.k.producer.Demo06ProducerTest    : [testSyncSend][发送编号：[1] 发送队列：[9]]
2019-12-07 18:48:45.867  INFO 31773 --- [           main] c.i.s.l.k.producer.Demo06ProducerTest    : [testSyncSend][发送编号：[1] 发送队列：[9]]
2019-12-07 18:48:45.869  INFO 31773 --- [           main] c.i.s.l.k.producer.Demo06ProducerTest    : [testSyncSend][发送编号：[1] 发送队列：[9]]
2019-12-07 18:48:45.870  INFO 31773 --- [           main] c.i.s.l.k.producer.Demo06ProducerTest    : [testSyncSend][发送编号：[1] 发送队列：[9]]
2019-12-07 18:48:45.871  INFO 31773 --- [           main] c.i.s.l.k.producer.Demo06ProducerTest    : [testSyncSend][发送编号：[1] 发送队列：[9]]
2019-12-07 18:48:45.872  INFO 31773 --- [           main] c.i.s.l.k.producer.Demo06ProducerTest    : [testSyncSend][发送编号：[1] 发送队列：[9]]
2019-12-07 18:48:45.873  INFO 31773 --- [           main] c.i.s.l.k.producer.Demo06ProducerTest    : [testSyncSend][发送编号：[1] 发送队列：[9]]
2019-12-07 18:48:45.875  INFO 31773 --- [           main] c.i.s.l.k.producer.Demo06ProducerTest    : [testSyncSend][发送编号：[1] 发送队列：[9]]
2019-12-07 18:48:45.876  INFO 31773 --- [           main] c.i.s.l.k.producer.Demo06ProducerTest    : [testSyncSend][发送编号：[1] 发送队列：[9]]
2019-12-07 18:48:45.877  INFO 31773 --- [           main] c.i.s.l.k.producer.Demo06ProducerTest    : [testSyncSend][发送编号：[1] 发送队列：[9]]

# Demo06Consumer 在线程编号为 18 中，顺序消费
2019-12-07 18:48:45.908  INFO 31773 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1}]
2019-12-07 18:48:45.911  INFO 31773 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1}]
2019-12-07 18:48:45.912  INFO 31773 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1}]
2019-12-07 18:48:45.912  INFO 31773 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1}]
2019-12-07 18:48:45.912  INFO 31773 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1}]
2019-12-07 18:48:45.912  INFO 31773 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1}]
2019-12-07 18:48:45.912  INFO 31773 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1}]
2019-12-07 18:48:45.912  INFO 31773 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1}]
2019-12-07 18:48:45.912  INFO 31773 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1}]
2019-12-07 18:48:45.912  INFO 31773 --- [ntainer#0-1-C-1] c.i.s.l.k.consumer.Demo06Consumer        : [onMessage][线程编号:18 消息内容：Demo06Message{id=1}]

😈 胖友认真看下艿艿在日志中的注释，理解下整个顺序消息的过程。

11. 事务消息

示例代码对应仓库：lab-03-kafka-demo-transaction 。

Kafka 内置提供事务消息的支持。对事务消息的概念不了解的胖友，可以看看《事务消息组件的套路》文章。

不过 Kafka 提供的并不是完整的的事务消息的支持，缺少了回查机制。关于这一点，刚推荐的文章也有讲到。目前，常用的分布式消息队列，只有 RocketMQ 提供了完整的事务消息的支持，具体的可以看看《芋道 Spring Boot 消息队列 RocketMQ 入门》的「9. 事务消息」小节，😈 暂时不拓展开来讲。

下面，我们开始本小节的示例。考虑到不污染上述的示例，我们新建一个 lab-03-kafka-demo-transaction 项目。

11.1 引入依赖

和 3.1 引入依赖」一致，见 pom.xml 文件。

11.2 应用配置文件

在 resources 目录下，创建 application.yaml 配置文件。配置如下：

spring:
  # Kafka 配置项，对应 KafkaProperties 配置类
  kafka:
    bootstrap-servers: 127.0.0.1:9092 # 指定 Kafka Broker 地址，可以设置多个，以逗号分隔
    # Kafka Producer 配置项
    producer:
      acks: all # 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 的序列化
      transaction-id-prefix: demo. # 事务编号前缀
    # Kafka Consumer 配置项
    consumer:
      auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring:
          json:
            trusted:
              packages: cn.iocoder.springboot.lab03.kafkademo.message
        isolation:
          level: read_committed # 读取已提交的消息
    # Kafka Consumer Listener 监听器配置
    listener:
      missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错

logging:
  level:
    org:
      springframework:
        kafka: ERROR # spring-kafka INFO 日志太多了，所以我们限制只打印 ERROR 级别
      apache:
        kafka: ERROR # kafka INFO 日志太多了，所以我们限制只打印 ERROR 级别

相比「3.2 应用配置文件」来说，修改或增加如下三个参数：
- 修改 spring.kafka.producer.acks=all 配置，不然在启动时会报 "Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence." 错误。因为，Kafka 的事务消息需要基于幂等性来实现，所以必须保证所有节点都写入成功。
- 增加 transaction-id-prefix=demo. 配置，事务编号的前缀。需要保证相同应用配置相同，不同应用配置不同。具体可以看看《How to choose Kafka transaction id for several applications》的讨论。
- 增加 spring.kafka.consumer.properties.isolation.level=read_committed 配置，Consumer 仅读取已提交的消息。😈 一定要配置！！！被坑惨了，当时以为自己的事务消息怎么就是不生效，原来少加了这个。

11.3 Demo07Producer

在 cn.iocoder.springboot.lab03.kafkademo.producer 包下，创建 Demo07Producer 类，它会使用 Kafka-Spring 封装提供的 KafkaTemplate ，实现发送事务消息。代码如下：

// Demo07Producer.java

@Component
public class Demo07Producer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    public String syncSendInTransaction(Integer id, Runnable runner) throws ExecutionException, InterruptedException {
        return kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<Object, Object, String>() {

            @Override
            public String doInOperations(KafkaOperations<Object, Object> kafkaOperations) {
                // 创建 Demo07Message 消息
                Demo07Message message = new Demo07Message();
                message.setId(id);
                try {
                    SendResult<Object, Object> sendResult = kafkaOperations.send(Demo07Message.TOPIC, message).get();
                    logger.info("[doInOperations][发送编号：[{}] 发送结果：[{}]]", id, sendResult);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }

                // 本地业务逻辑... biubiubiu
                runner.run();

                // 返回结果
                return "success";
            }

        });
    }

}

使用 kafkaTemplate 提交的 #executeInTransaction(OperationsCallback<K, V, T> callback) 模板方法，实现在 Kafka 事务中，执行自定义 KafkaOperations.OperationsCallback 操作。
- 在 #executeInTransaction(...) 方法中，我们可以通过 KafkaOperations 来执行发送消息等 Kafka 相关的操作，也可以执行自己的业务逻辑。
- 在 #executeInTransaction(...) 方法的开始，它会自动动创建 Kafka 的事务；然后执行我们定义的 KafkaOperations 的逻辑；如果成功，则提交 Kafka 事务；如果失败，则回滚 Kafka 事务。
另外，我们定义了一个 runner 参数，用于表示本地业务逻辑~

注意，如果 Kafka Producer 开启了事务的功能，则所有发送的消息，都必须处于 Kafka 事务之中，否则会抛出 " No transaction is in process; possible solutions: run the template operation within the scope of a template.executeInTransaction() operation, start a transaction with @Transactional before invoking the template method, run in a transaction started by a listener container when consuming a record" 异常。

所以，如果胖友的业务中，即存在需要事务的情况，也存在不需要事务的情况，需要分别定义两个 KafkaTemplate（Kafka Producer）。

11.4 Demo07Consumer

在 cn.iocoder.springboot.lab03.kafkademo.consumer 包下，创建 Demo07Consumer 类，消费消息。代码如下：

// Demo07Consumer.java

@Component
public class Demo07Consumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = Demo07Message.TOPIC,
            groupId = "demo07-consumer-group-" + Demo07Message.TOPIC)
    public void onMessage(Demo07Message message) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

}

和「3.6 Demo1Consumer.java」一致，差别只在于消费 Topic 为 DEMO_07 的消息。

11.5 简单测试

创建 Demo07ProducerTest 测试类，编写单元测试方法，调用 Demo07Producer 发送事务消息的方式。代码如下：

// Demo07ProducerTest.java

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class Demo07ProducerTest {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private Demo07Producer producer;

    @Test
    public void testSyncSendInTransaction() throws ExecutionException, InterruptedException {
        int id = (int) (System.currentTimeMillis() / 1000);
        producer.syncSendInTransaction(id, new Runnable() {

            @Override
            public void run() {
                logger.info("[run][我要开始睡觉了]");
                try {
                    Thread.sleep(10 * 1000L);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                logger.info("[run][我睡醒了]");
            }

        });

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }

}

故意创建一个执行逻辑为 sleep 10 秒的 Runnable 对象，来让我们测试验证，事务消息在提交后，才会被 Consumer 消费到。

执行 #testSyncSendInTransaction() 单元测试，输出日志如下：

# Producer 同步发送消息成功。
2019-12-07 21:10:20.496  INFO 37455 --- [           main] c.i.s.l.k.producer.Demo07Producer        : [doInOperations][发送编号：[1575724220] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=DEMO_07, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 55, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo07Message{id=1575724220}, timestamp=null), recordMetadata=DEMO_07-0@14]]]

# 故意 sleep 10 秒，延迟事务消息的提交
2019-12-07 21:10:20.496  INFO 37455 --- [           main] c.i.s.l.k.producer.Demo07ProducerTest    : [run][我要开始睡觉了]
2019-12-07 21:10:30.500  INFO 37455 --- [           main] c.i.s.l.k.producer.Demo07ProducerTest    : [run][我睡醒了]

# 在事务消息提交之后，事务消息才被 Consumer 消费到
2019-12-07 21:10:30.558  INFO 37455 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo07Consumer        : [onMessage][线程编号:16 消息内容：Demo07Message{id=1575724220}]

成功观察到，发送的事务消息，在提交之后才被 Consumer 消费到。

11.6 集成到 Spring Transaction 体系

Spring-Kafka 提供了对 Spring Transaction 的集成，所以在实际开发中，我们只需要配合使用 @Transactional 注解，来声明事务即可，而无需使用 KafkaTemplate 提供的 #executeInTransaction(...) 模板方法。😈 是不是便捷很多呢!

具体的使用示例，艿艿就暂时不提供了，感兴趣的胖友可以看看《使用Kafka 事务的两种方式》文章。

12. 消费进度的提交机制

示例代码对应仓库：lab-03-kafka-demo-ack 。

原生 Kafka Consumer 消费端，有两种消费进度提交的提交机制：

【默认】自动提交，通过配置 enable.auto.commit=true ，每过 auto.commit.interval.ms 时间间隔，都会自动提交消费消费进度。而提交的时机，是在 Consumer 的 #poll(...) 方法的逻辑里完成，在每次从 Kafka Broker 拉取消息时，会检查是否到达自动提交的时间间隔，如果是，那么就会提交上一次轮询拉取的位置。
手动提交，通过配置 enable.auto.commit=false ，后续通过 Consumer 的 #commitSync(...) 或 #commitAsync(...) 方法，同步或异步提交消费进度。

Spring-Kafka Consumer 消费端，提供了更丰富的消费者进度的提交机制，更加灵活。当然，也是分成自动提交和手动提交两个大类。在 AckMode 枚举类中，可以看到每一种具体的方式。代码如下：

// ContainerProperties#AckMode.java

public enum AckMode {

// ========== 自动提交 ==========

	/**
	 * Commit after each record is processed by the listener.
	 */
	RECORD, // 每条消息被消费完成后，自动提交

	/**
	 * Commit whatever has already been processed before the next poll.
	 */
	BATCH, // 每一次消息被消费完成后，在下次拉取消息之前，自动提交

	/**
	 * Commit pending updates after
	 * {@link ContainerProperties#setAckTime(long) ackTime} has elapsed.
	 */
	TIME, // 达到一定时间间隔后，自动提交。
	      // 不过要注意，它并不是一到就立马提交，如果此时正在消费某一条消息，需要等这条消息被消费完成，才能提交消费进度。

	/**
	 * Commit pending updates after
	 * {@link ContainerProperties#setAckCount(int) ackCount} has been
	 * exceeded.
	 */
	COUNT, // 消费成功的消息数到达一定数量后，自动提交。
	      // 不过要注意，它并不是一到就立马提交，如果此时正在消费某一条消息，需要等这条消息被消费完成，才能提交消费进度。

	/**
	 * Commit pending updates after
	 * {@link ContainerProperties#setAckCount(int) ackCount} has been
	 * exceeded or after {@link ContainerProperties#setAckTime(long)
	 * ackTime} has elapsed.
	 */
	COUNT_TIME, // TIME 和 COUNT 的结合体，满足任一都会自动提交。

// ========== 手动提交 ==========

	/**
	 * User takes responsibility for acks using an
	 * {@link AcknowledgingMessageListener}.
	 */
	MANUAL, // 调用时，先标记提交消费进度。等到当前消息被消费完成，然后在提交消费进度。

	/**
	 * User takes responsibility for acks using an
	 * {@link AcknowledgingMessageListener}. The consumer
	 * immediately processes the commit.
	 */
	MANUAL_IMMEDIATE, // 调用时，立即提交消费进度。

}

看下每种方式，艿艿都添加了注释哟。

那么，既然现在存在原生 Kafka 和 Spring-Kafka 提供的两种消费进度的提交机制，我们应该怎么配置呢？

使用原生 Kafka 的方式，通过配置 spring.kafka.consumer.enable-auto-commit=true 。然后，通过 spring.kafka.consumer.auto-commit-interval 设置自动提交的频率。
使用 Spring-Kafka 的方式，通过配置 spring.kafka.consumer.enable-auto-commit=false 。然后通过 spring.kafka.listener.ack-mode 设置具体模式。另外，还有 spring.kafka.listener.ack-time 和 spring.kafka.listener.ack-count 可以设置自动提交的时间间隔和消息条数。

默认什么都不配置的情况下，使用 Spring-Kafka 的 BATCH 模式：每一次消息被消费完成后，在下次拉取消息之前，自动提交。

下面，我们开始本小节的示例，实现一个手动提交消息进度的消费者。考虑到不污染上述的示例，我们新建一个 lab-03-kafka-demo-ack 项目。

12.1 引入依赖

和「3.1 引入依赖」」一致，见 pom.xml 文件

12.2 应用配置文件

在 resources 目录下，创建 application.yaml 配置文件。配置如下：

spring:
  # Kafka 配置项，对应 KafkaProperties 配置类
  kafka:
    bootstrap-servers: 127.0.0.1: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 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      enable-auto-commit: false # 使用 Spring-Kafka 的消费进度的提交机制
      properties:
        spring:
          json:
            trusted:
              packages: cn.iocoder.springboot.lab03.kafkademo.message
    # Kafka Consumer Listener 监听器配置
    listener:
      missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错
      ack-mode: MANUAL

logging:
  level:
    org:
      springframework:
        kafka: ERROR # spring-kafka INFO 日志太多了，所以我们限制只打印 ERROR 级别
      apache:
        kafka: ERROR # kafka INFO 日志太多了，所以我们限制只打印 ERROR 级别

相比「3.2 应用配置文件」来说，增加如下两个参数：
- 添加 spring.kafka.consumer.enable-auto-commit=false 配置，使用 Spring-Kafka 的消费进度的提交机制。😈 设计情况下，不添加该配置项也是可以的，因为 false 是默认值。
- 添加 spring.kafka.listener.ack-mode=MANUAL 配置，使用 MANUAL 模式：调用时，先标记提交消费进度。等到当前消息被消费完成，然后在提交消费进度。

12.3 Demo08Message

在 cn.iocoder.springboot.lab03.kafkademo.message 包下，创建 Demo08Message 消息类，提供给当前示例使用。代码如下：

// Demo04Message.java

public class Demo08Message {

    public static final String TOPIC = "DEMO_08";

    /**
     * 编号
     */
    private Integer id;

    // ... 省略 set/get/toString 方法

}

TOPIC 静态属性，我们设置该消息类对应 Topic 为 "DEMO_08" 。

12.4 Demo08Producer

在 cn.iocoder.springboot.lab03.kafkademo.producer 包下，创建 Demo08Producer 类，它会使用 Kafka-Spring 封装提供的 KafkaTemplate ，同步发送消息。代码如下：

// Demo08Producer.java

@Component
public class Demo08Producer {

    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;

    public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
        // 创建 Demo08Message 消息
        Demo08Message message = new Demo08Message();
        message.setId(id);
        // 同步发送消息
        return kafkaTemplate.send(Demo08Message.TOPIC, message).get();
    }
    
}

和「3.5 Demo01Producer」的同步发送消息的代码是一致的，就是换成了 Demo08Message 。

12.5 Demo08Consumer

在 cn.iocoder.springboot.lab03.kafkademo.consumer 包下，创建 Demo08Consumer 类，消费消息。代码如下：

// Demo08Consumer.java

@Component
public class Demo08Consumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = Demo08Message.TOPIC,
            groupId = "demo08-consumer-group-" + Demo08Message.TOPIC)
    public void onMessage(Demo08Message message, Acknowledgment acknowledgment) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
        // 提交消费进度
        if (message.getId() % 2 == 1) {
            acknowledgment.acknowledge();
        }
    }

}

在消费方法上，我们增加了第二个方法参数，类型为 Acknowledgment 类。通过调用其 #acknowledge() 方法，可以提交当前消息的 Topic 的 Partition 的消费进度。
在消费逻辑中，我们故意只提交消费的消息的 Demo08Message.id 为奇数的消息。😈 这样，我们只需要发送一条 id=1 ，一条 id=2 的消息，如果第二条的消费进度没有被提交，就可以说明手动提交消费进度成功。

12.6 简单测试

创建 Demo08ProducerTest 测试类，编写单元测试方法，测试手动提交消费进度。代码如下：

// Demo08ProducerTest.java

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class Demo08ProducerTest {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private Demo08Producer producer;

    @Test
    public void testSyncSend() throws ExecutionException, InterruptedException {
        for (int id = 1; id <= 2; id++) {
            SendResult result = producer.syncSend(id);
            logger.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);
        }

        // 阻塞等待，保证消费
        new CountDownLatch(1).await();
    }

}

执行 #testSyncSend() 单元测试，输出日志如下：

// Producer 同步发送 2 条消息成功
2019-12-07 23:41:20.914  INFO 43412 --- [           main] c.i.s.l.k.producer.Demo08ProducerTest    : [testSyncSend][发送编号：[1] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=DEMO_08, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 56, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo08Message{id=1}, timestamp=null), recordMetadata=DEMO_08-0@0]]]
2019-12-07 23:41:20.916  INFO 43412 --- [           main] c.i.s.l.k.producer.Demo08ProducerTest    : [testSyncSend][发送编号：[2] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=DEMO_08, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103, 98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 56, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo08Message{id=2}, timestamp=null), recordMetadata=DEMO_08-0@1]]]

// Demo08Consumer 消费 2 条消息成功
2019-12-07 23:41:21.006  INFO 43412 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo08Consumer        : [onMessage][线程编号:16 消息内容：Demo08Message{id=1}]
2019-12-07 23:41:21.006  INFO 43412 --- [ntainer#0-0-C-1] c.i.s.l.k.consumer.Demo08Consumer        : [onMessage][线程编号:16 消息内容：Demo08Message{id=2}]

此时，如果我们使用 Kafka Manager 来查看 "DEMO_08" 的消费者列表：

第 2 条消息的消费进度，未被提交，符合预期~

666. 彩蛋

Spring-Kafka 提供的配置项非常丰富，艿艿在本文只覆盖了相对常用的部分，所以胖友可以在有需要的时候，看看《Spring-Kafka 生产者消费者配置详解》和《Kafka Producer配置解读》文章。

因为艿艿个人在生产环境下，主要是使用 RocketMQ 作为消息队列。如果有写的不正确的地方，辛苦胖友帮忙指正。这里额外在推荐一些 Kafka 不错的内容：

《深入理解 Kafka 核心设计与实践原理》

感谢厮大（本书作者）在艿艿写本文时，各种智障的问题的指导。

所以，本文有任何内容的错误，都是厮大教的不对。
《Kafka Exactly-Once 之事务性实》
《阿里云 Kafka 版 —— 最佳实践》
《针对 Spring-Kafka 的 Consumer 端上的使用分析总结》

最后弱弱的说一下：

对于业务平台来说，确实觉得 RocketMQ 更加好用。
对于数据平台来说，应该是 Kafka 更加好用。