kafka学习笔记(二)：Spring-Kafka和更多使用实例

阅读量

参考文章

1 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(功能特性)

1.1 引入依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
<!-- 引入 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>

1.2 应用配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
spring:
  # Kafka 配置项，对应 KafkaProperties 配置类
  kafka:
    bootstrap-servers: 39.97.209.29:9094 # 指定 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: com.example.kafka.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 级别
server:
  port: 8080

  • 在 spring.kafka 配置项，设置 Kafka 的配置，对应 KafkaProperties 配置类。

  • Spring Boot 提供的 KafkaAutoConfiguration 自动化配置类，实现 Kafka 的自动配置，创建相应的 Producer 和 Consumer 。

  • spring.kafka.bootstrap-servers 配置项，设置 Kafka Broker 地址。如果多个，使用逗号分隔。

  • spring.kafka.producer配置项

    • value-serializer 配置，使用了 Spring-Kafka 提供的 JsonSerializer 序列化类
    • 其它配置，一般默认即可。

  • spring.kafka.consumer配置项

    • value-serializer 配置，使用了 Spring-Kafka 提供的 JsonDeserializer 反序列化类
    • properties.spring.json.trusted.packages 配置，配置信任 com.example.kafka.message 包下的 Message 类们。因为 JsonDeserializer 在反序列化消息时，考虑到安全性，只反序列化成信任的 Message 类。

1.3 Demo01Message

在com.example.kafka.message包下创建Demo01Message类提供给当前示例使用。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
	// Demo01Message.java

public class Demo01Message {

    public static final String TOPIC = "DEMO_01";

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

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

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

1.4 Demo01Producer

在com.example.kafka.producer包下创建Demo01Producer，使用 Kafka-Spring 封装提供的 KafkaTemplate ，实现三种发送消息的方式。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
	// 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);
    }

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

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

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

1.5 Demo01Consumer

在com.example.kafka.consumer包下创建Demo01Consumer消费消息。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
@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 对应的消息类即可。这里，我们使用了Demo01Message 。
  • 虽然@KafkaListener 注解是方法级别的，但是建议一个类对应一个方法消费消息。

1.6 Demo01AConsumer

在com.example.kafka.consumer包下创建Demo01AConsumer类，消费消息。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 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);
    }
    
}
  • 整体和 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 类。

1.7 简单测试

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@RunWith(SpringRunner.class)
@SpringBootTest(classes = KafkaApplication.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() 方法，测试同步发送消息。控制台输出如下：

1
2
3
4
5
6
7
8
# Producer 同步发送消息成功。注意 __TypeId__
2022-12-28 15:25:23.738  INFO 38792 --- [           main] com.example.kafka.Demo01ProducerTest     : [testSyncSend][发送编号：[1672212323] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=DEMO_01, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 101, 120, 97, 109, 112, 108, 101, 46, 107, 97, 102, 107, 97, 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=1672212323}, timestamp=null), recordMetadata=DEMO_01-0@8]]]

# Demo01AConsumer 消费了一次该消息
2022-12-28 15:25:23.849  INFO 38792 --- [ntainer#0-0-C-1] c.e.kafka.consumer.Demo01AConsumer       : [onMessage][线程编号:17 消息内容：ConsumerRecord(topic = DEMO_01, partition = 0, leaderEpoch = 0, offset = 8, CreateTime = 1672212323562, serialized key size = -1, serialized value size = 17, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = Demo01Message{id=1672212323})]

# Demo01Consumer 消费了一次该消息
2022-12-28 15:25:23.853  INFO 38792 --- [ntainer#1-0-C-1] c.example.kafka.consumer.Demo01Consumer  : [onMessage][线程编号:19 消息内容：Demo01Message{id=1672212323}]
  • 通过日志我们可以看到，我们发送的消息，分别被 Demo01AConsumer 和 Demo01Consumer 两个消费者（消费者分组）都消费了一次。
  • 同时，两个消费者在不同的线程中，消费了这条消息。

不关闭 #testSyncSend() 单元测试方法，模拟每个消费者集群有多个 Consumer 节点。

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

1
2
3
4
5
6
7
8
#这条日志出现在testASyncSend窗口
2022-12-28 15:28:36.399  INFO 40208 --- [ad | producer-1] com.example.kafka.Demo01ProducerTest     : [testASyncSend][发送编号：[1672212516] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_01, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 101, 120, 97, 109, 112, 108, 101, 46, 107, 97, 102, 107, 97, 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=1672212516}, timestamp=null), recordMetadata=DEMO_01-0@9]]]

#下面两条出现在testSyncSend窗口
2022-12-28 15:28:36.398  INFO 38792 --- [ntainer#0-0-C-1] c.e.kafka.consumer.Demo01AConsumer       : [onMessage][线程编号:17 消息内容：ConsumerRecord(topic = DEMO_01, partition = 0, leaderEpoch = 0, offset = 9, CreateTime = 1672212516309, serialized key size = -1, serialized value size = 17, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = Demo01Message{id=1672212516})]

2022-12-28 15:28:36.421  INFO 38792 --- [ntainer#1-0-C-1] c.example.kafka.consumer.Demo01Consumer  : [onMessage][线程编号:19 消息内容：Demo01Message{id=1672212516}]

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

1.8 @KafkaListener

在Demo01Consumer中使用了@KafkaListener注解设置每个 Kafka 消费者 Consumer 的消息监听器的配置。

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
/**
 * 监听的 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 注解的不常用属性如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
/**
 * 唯一标识
 *  
 * 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";

2 批量发送消息

Kafka 提供了一个 RecordAccumulator 消息收集器，将发送给相同 Topic 的相同 Partition 分区的消息们，“偷偷”收集在一起，当满足条件时候，一次性批量发送提交给 Kafka Broker 。如下是三个条件，满足任一即会批量发送：

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

2.1 增加配置项

1
2
3
4
5
6
7
8
9
10
11
spring:
  # Kafka 配置项，对应 KafkaProperties 配置类
  kafka:
    bootstrap-servers: 39.97.209.29:9094 # 指定 Kafka Broker 地址，可以设置多个，以逗号分隔
    # Kafka Producer 配置项
    producer:
      batch-size: 16384 # 每次批量发送消息的最大数量
      buffer-memory: 33554432 # 每次批量发送消息的最大内存
      properties:
        linger:
          ms: 30000 # 批处理延迟时间上限。这里配置为 30 * 1000 ms 过后，不管是否消息数量是否到达 batch-size 或者消息大小到达 buffer-memory 后，都直接发送一次请求。

额外三个参数，就是我们说的 Producer 批量发送的三个条件：

  • spring.kafka.producer.batch-size
  • spring.kafka.producer.buffer-memory
  • spring.kafka.producer.properties.linger.ms

2.2 Demo02Message

在 com.example.kafka.message 包下，创建Demo02Message消息类，提供给当前示例使用。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
package com.example.kafka.message;

// Demo02Message.java
public class Demo02Message {

    public static final String TOPIC = "DEMO_012";

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

	 // ... 省略 set/get/toString 方法
}
  • TOPIC 静态属性，我们设置该消息类对应 Topic 为 "DEMO_02" 。

2.3 Demo02Producer

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 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);
    }

}
  • 看起来和我们在1.4 Demo01Producer提供的异步发送消息的方法，除了换成了 Demo02Message 消息对象，其它都是一模一样的。
  • 因为我们发送的消息 Topic 是自动创建的，所以其 Partition 分区大小是 1 。这样，就能保证我每次调用这个方法，满足批量发送消息的一个前提，相同 Topic 的相同 Partition 分区的消息们。

2.4 Demo02Consumer

在com.example.kafka.consumer包下，创建Demo02Consumer类，消费消息。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 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);
    }

}
  • 和1.5 Demo01Consumer基本一致，除了是不同的消费者分组，消费了不同的 Topic 。

2.5 测试

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
// Demo02ProducerTest.java

@RunWith(SpringRunner.class)
@SpringBootTest(classes = KafkaApplication.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() 方法，测试批量发送消息。控制台输出如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 打印 testASyncSend 方法开始执行的日志
2022-12-28 16:09:25.604  INFO 21372 --- [           main] com.example.kafka.Demo02ProducerTest     : [testASyncSend][开始执行]

# 30 秒后，满足批量消息的最大等待时长，所以 3 条消息被 Producer 批量发送。
# 因此我们配置的是 acks=1 ，需要等待发送成功后，才会回调 ListenableFutureCallback 的方法。
2022-12-28 16:09:55.943  INFO 21372 --- [ad | producer-1] com.example.kafka.Demo02ProducerTest     : [testASyncSend][发送编号：[1672214965] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_012, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 101, 120, 97, 109, 112, 108, 101, 46, 107, 97, 102, 107, 97, 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=Demo02Message{id=1672214965}, timestamp=null), recordMetadata=DEMO_012-0@0]]]
2022-12-28 16:09:55.944  INFO 21372 --- [ad | producer-1] com.example.kafka.Demo02ProducerTest     : [testASyncSend][发送编号：[1672214975] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_012, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 101, 120, 97, 109, 112, 108, 101, 46, 107, 97, 102, 107, 97, 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=Demo02Message{id=1672214975}, timestamp=null), recordMetadata=DEMO_012-0@1]]]
2022-12-28 16:09:55.944  INFO 21372 --- [ad | producer-1] com.example.kafka.Demo02ProducerTest     : [testASyncSend][发送编号：[1672214985] 发送成功，结果为：[SendResult [producerRecord=ProducerRecord(topic=DEMO_012, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 101, 120, 97, 109, 112, 108, 101, 46, 107, 97, 102, 107, 97, 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=Demo02Message{id=1672214985}, timestamp=null), recordMetadata=DEMO_012-0@2]]]

# 因为 Producer 批量发送完成，所以 Demo02Consumer 消费到消息
2022-12-28 16:09:55.979  INFO 21372 --- [ntainer#2-0-C-1] c.example.kafka.consumer.Demo02Consumer  : [onMessage][线程编号:17 消息内容：Demo02Message{id=1672214965}]
2022-12-28 16:09:55.979  INFO 21372 --- [ntainer#2-0-C-1] c.example.kafka.consumer.Demo02Consumer  : [onMessage][线程编号:17 消息内容：Demo02Message{id=1672214975}]
2022-12-28 16:09:55.980  INFO 21372 --- [ntainer#2-0-C-1] c.example.kafka.consumer.Demo02Consumer  : [onMessage][线程编号:17 消息内容：Demo02Message{id=1672214985}]

3 定时消息

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

参考《基于 Kafka 的定时消息/任务服》

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

4 消费重试

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

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

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

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

每条消息的失败重试，是可以配置一定的间隔时间。参考下面示例

4.1 KafkaConfiguration

在包com.example.kafka.config下创建KafkaConfiguration配置类如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 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<3> 会将 调用 Kafka Consumer 的 #seek(TopicPartition partition, long offset) 方法，将 Consumer 对于该消息对应的 TopicPartition 分区的本地进度设置成该消息的位置。这样，Consumer 在下次从 Kafka Broker 拉取消息的时候，又能重新拉取到这条消费失败的消息。

  • 另一方面，Spring-Kafka 使用 FailedRecordTracker 对每个 Topic 的每个 TopicPartition 消费失败次数进行计数，这样相当于对该 TopicPartition 的第一条消费失败的消息的消费失败次数进行计数。

    • 在 FailedRecordTracker 中，会调用 BackOff<2>来进行计算，该消息的下一次重新消费的时间，通过 Thread#sleep(...) 方法，实现重新消费的时间间隔。

  • 有一点需要注意，FailedRecordTracker 提供的计数是客户端级别的，重启 JVM 应用后，计数是会丢失的。所以，如果想要计数进行持久化，需要自己自行实现下 FailedRecordTracker 类，通过 ZooKeeper 存储计数。

4.2 Demo04Message

在包com.example.kafka.message下创建Demo04Message如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// Demo04Message.java

public class Demo04Message {

    public static final String TOPIC = "DEMO_04";

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

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

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

4.3 Demo04Producer

在com.example.kafka.producer包下创建Demo04Producer如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 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();
    }

}

4.4 Demo04Consumer

在com.example.kafka.producer包下创建Demo04Consumer如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 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("我就是故意抛出一个异常");
    }

}

4.5 简单测试

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 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方法结果如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#发送成功
2023-01-06 14:51:10.857  INFO 18592 --- [           main] com.example.kafka.Demo04ProducerTest     : [testSyncSend][发送编号：[1672987840] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=DEMO_04, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 101, 120, 97, 109, 112, 108, 101, 46, 107, 97, 102, 107, 97, 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=1672987840), timestamp=null), recordMetadata=DEMO_04-0@1]]]

#首次消费失败
2023-01-06 14:51:10.900  INFO 18592 --- [ntainer#3-0-C-1] c.example.kafka.consumer.Demo04Consumer  : [onMessage][线程编号:19 消息内容：Demo04Message(id=1672987840)]
2023-01-06 14:51:20.916 ERROR 18592 --- [ntainer#3-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception

#第一次重试消费失败
2023-01-06 14:51:20.945  INFO 18592 --- [ntainer#3-0-C-1] c.example.kafka.consumer.Demo04Consumer  : [onMessage][线程编号:19 消息内容：Demo04Message(id=1672987840)]
2023-01-06 14:51:30.952 ERROR 18592 --- [ntainer#3-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception

#第二次重试消费失败 
2023-01-06 14:51:30.962  INFO 18592 --- [ntainer#3-0-C-1] c.example.kafka.consumer.Demo04Consumer  : [onMessage][线程编号:19 消息内容：Demo04Message(id=1672987840)]
2023-01-06 14:51:40.974 ERROR 18592 --- [ntainer#3-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception

#第三次重试消费失败
2023-01-06 14:51:41.034  INFO 18592 --- [ntainer#3-0-C-1] c.example.kafka.consumer.Demo04Consumer  : [onMessage][线程编号:19 消息内容：Demo04Message(id=1672987840)]
    
# 这次不会打印异常日志，直接发到死信队列

5 广播消费

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

  • 不过 Kafka 并不直接提供内置的广播消费的功能。使每个 Consumer 独有一个 Consumer Group ，从而保证都能接收到全量的消息。

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

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

5.1 修改应用配置

和2.1 增加配置项中修改配置项 spring.kafka.consumer.auto-offset-reset=latest 。因为在广播订阅下，一般情况下，无需消费历史的消息，而是从订阅的 Topic 的队列的尾部开始消费即可，所以配置为 latest 。

5.2 Demo05Message

在 com.example.kafka.message 包下，创建 Demo05Message 消息类，提供给当前示例使用。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// Demo05Message.java

public class Demo05Message {

    public static final String TOPIC = "DEMO_05";

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

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

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

5.3 Demo05Producer

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 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();
    }

}
  • 和1.4 Demo01Producer的同步发送消息的代码是一致的，就是换成了 Demo05Message 。

5.4 Demo05Consumer

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// Demo04Consumer.java

@Component
public class Demo05Consumer {

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

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

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

5.5 简单测试

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@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) 方法，同步发送了一条消息。控制台输出如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#### testSyncSend 方法对应的控制台 ####

# Producer 同步发送消息成功
2023-01-12 10:37:58.938  INFO 23704 --- [           main] com.example.kafka.Demo05ProducerTest     : [testSyncSend][发送编号：[1673491048] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=DEMO_05, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 101, 120, 97, 109, 112, 108, 101, 46, 107, 97, 102, 107, 97, 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=1673491048), timestamp=null), recordMetadata=DEMO_05-0@0]]]

    
# Demo05Consumer 消费了该消息
2023-01-12 10:37:59.002  INFO 23704 --- [ntainer#4-0-C-1] c.example.kafka.consumer.Demo05Consumer  : [onMessage][线程编号:21 消息内容：Demo05Message(id=1673491048)]


### test 方法对应的控制台 ####
    
# Demo05Consumer 也消费了该消息
2023-01-12 10:37:59.064  INFO 27104 --- [ntainer#4-0-C-1] c.example.kafka.consumer.Demo05Consumer  : [onMessage][线程编号:21 消息内容：Demo05Message(id=1673491048)]

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

6 并发消费

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

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

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

例如：

  • 创建一个 Topic 为 "DEMO_06" ，并且设置其 Partition 分区数为 10 。
  • 创建一个 Demo06Consumer 类，并在其消费方法上，添加 @KafkaListener(concurrency=2) 注解。
  • 启动项目。Spring-Kafka 会根据 @KafkaListener(concurrency=2) 注解，创建 2 个 Kafka Consumer 。每个 Kafka Consumer 会被单独分配到一个线程中，进行拉取消息，消费消息。
  • 之后，Kafka Broker 会将 Topic 为 "DEMO_06" 分配给创建的 2 个 Kafka Consumer 各 5 个 Partition 。
  • 因为 @KafkaListener(concurrency=2) 注解，创建 2 个 Kafka Consumer ，在各自的线程中，拉取各自的 Topic 为 "DEMO_06" 的 Partition 的消息，各自串行消费。从而，实现多线程的并发消费。

6.1 引入依赖

和 1.1 引入依赖 一致

6.2 应用配置文件

和 1.2 应用配置文件 一致

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

6.3 Demo06Message

在 com.example.kafka.message 包下，创建 Demo06Message 消息类，提供给当前示例使用。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// Demo06Message.java

public class Demo06Message {

    public static final String TOPIC = "DEMO_06";

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

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

}

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

1
$ ./kafka-topics.sh --create --topic DEMO_06 --bootstrap-server localhost:9094 --replication-factor 1 --partitions 10

6.4 Demo06Producer

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 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();
    }

}
  • 和1.1 Demo01Producer的同步发送消息的代码是一致的，就是换成了 Demo06Message 。

6.5 Demo06Consumer

在 com.example.kafka.consumer 类，消费消息。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// Demo06Consumer.java

@Component
public class Demo06Consumer {

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

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

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

6.6 简单测试

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@RunWith(SpringRunner.class)
@SpringBootTest(classes = KafkaApplication.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 测试方法结果如下

1
2
3
4
5
6
7
8
9
10
11
2023-01-16 13:19:26.220  INFO 6548 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1673846366)]
2023-01-16 13:19:26.220  INFO 6548 --- [ntainer#5-0-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:23 消息内容：Demo06Message(id=1673846365)]
2023-01-16 13:19:26.245  INFO 6548 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1673846366)]
2023-01-16 13:19:26.245  INFO 6548 --- [ntainer#5-0-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:23 消息内容：Demo06Message(id=1673846366)]
2023-01-16 13:19:26.245  INFO 6548 --- [ntainer#5-0-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:23 消息内容：Demo06Message(id=1673846366)]
2023-01-16 13:19:26.245  INFO 6548 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1673846366)]
2023-01-16 13:19:26.246  INFO 6548 --- [ntainer#5-0-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:23 消息内容：Demo06Message(id=1673846366)]
2023-01-16 13:19:26.246  INFO 6548 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1673846366)]
2023-01-16 13:19:26.246  INFO 6548 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1673846366)]
2023-01-16 13:19:26.261  INFO 6548 --- [ntainer#5-0-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:23 消息内容：Demo06Message(id=1673846366)]

可以看到有两个线程在消费消息，此时，使用 Kafka Manager 来查看 "DEMO_06" 的消费者列表：也可以看到有两个消费者各负责消费了5个分区

image-20230116132514290

7 顺序消息

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

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

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

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

下面示例，我们根据6. 并发消费项目改造。

7.1 Demo06Producer

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

1
2
3
4
5
6
7
8
9
10
// 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 中。

7.2 简单测试

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

1
2
3
4
5
6
7
8
9
10
11
12
13
// 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() 单元测试，输出日志如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
2023-01-18 14:44:40.410  INFO 51476 --- [           main] com.example.kafka.Demo06ProducerTest     : [testSyncSend][发送编号：[1] 发送队列：[9]]
2023-01-18 14:44:40.423  INFO 51476 --- [           main] com.example.kafka.Demo06ProducerTest     : [testSyncSend][发送编号：[1] 发送队列：[9]]
2023-01-18 14:44:40.439  INFO 51476 --- [           main] com.example.kafka.Demo06ProducerTest     : [testSyncSend][发送编号：[1] 发送队列：[9]]
2023-01-18 14:44:40.454  INFO 51476 --- [           main] com.example.kafka.Demo06ProducerTest     : [testSyncSend][发送编号：[1] 发送队列：[9]]
2023-01-18 14:44:40.466  INFO 51476 --- [           main] com.example.kafka.Demo06ProducerTest     : [testSyncSend][发送编号：[1] 发送队列：[9]]
2023-01-18 14:44:40.479  INFO 51476 --- [           main] com.example.kafka.Demo06ProducerTest     : [testSyncSend][发送编号：[1] 发送队列：[9]]
2023-01-18 14:44:40.507  INFO 51476 --- [           main] com.example.kafka.Demo06ProducerTest     : [testSyncSend][发送编号：[1] 发送队列：[9]]
2023-01-18 14:44:40.533  INFO 51476 --- [           main] com.example.kafka.Demo06ProducerTest     : [testSyncSend][发送编号：[1] 发送队列：[9]]
2023-01-18 14:44:40.546  INFO 51476 --- [           main] com.example.kafka.Demo06ProducerTest     : [testSyncSend][发送编号：[1] 发送队列：[9]]
2023-01-18 14:44:40.557  INFO 51476 --- [           main] com.example.kafka.Demo06ProducerTest     : [testSyncSend][发送编号：[1] 发送队列：[9]]

2023-01-18 14:44:40.642  INFO 51476 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1)]
2023-01-18 14:44:40.643  INFO 51476 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1)]
2023-01-18 14:44:40.644  INFO 51476 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1)]
2023-01-18 14:44:40.644  INFO 51476 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1)]
2023-01-18 14:44:40.644  INFO 51476 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1)]
2023-01-18 14:44:40.645  INFO 51476 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1)]
2023-01-18 14:44:40.645  INFO 51476 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1)]
2023-01-18 14:44:40.675  INFO 51476 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1)]
2023-01-18 14:44:40.675  INFO 51476 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1)]
2023-01-18 14:44:40.675  INFO 51476 --- [ntainer#5-1-C-1] c.example.kafka.consumer.Demo06Consumer  : [onMessage][线程编号:25 消息内容：Demo06Message(id=1)]

8 事务消息

Kafka 内置提供事务消息的支持。对事务消息的概念参考《事务消息组件的套路》 文章。

不过 Kafka 提供的并不是完整的的事务消息的支持，缺少了回查机制。关于这一点，刚推荐的文章也有讲到。目前，常用的分布式消息队列，只有 RocketMQ 提供了完整的事务消息的支持，参考《Spring Boot 消息队列 RocketMQ 入门》 的「9. 事务消息」小节。

8.1 引入依赖

和1.1 引入依赖一致

8.2 应用配置文件

application.yaml配置内容如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
spring:
  # Kafka 配置项，对应 KafkaProperties 配置类
  kafka:
    bootstrap-servers: 39.97.209.29:9094 # 指定 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:  com.example.kafka.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 级别
  • 相比1.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 仅读取已提交的消息。

8.3 Demo07Producer

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
// 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）。

8.4 Demo07Consumer

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 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);
    }

}
  • 和1.6 Demo1Consumer.java 一致，差别只在于消费 Topic 为 DEMO_07 的消息。

8.5 简单测试

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
// 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() 单元测试，输出日志如下：

1
2
3
4
5
6
7
8
9
10
# Producer 同步发送消息成功。
2023-01-18 15:37:24.161  INFO 51896 --- [           main] c.example.kafka.producer.Demo07Producer  : [doInOperations][发送编号：[1674027443] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=DEMO_07, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 101, 120, 97, 109, 112, 108, 101, 46, 107, 97, 102, 107, 97, 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=1674027443), timestamp=null), recordMetadata=DEMO_07-0@8]]]

# 故意 sleep 10 秒，延迟事务消息的提交
2023-01-18 15:37:24.162  INFO 51896 --- [           main] com.example.kafka.Demo07ProducerTest     : [run][我要开始睡觉了]
2023-01-18 15:37:34.172  INFO 51896 --- [           main] com.example.kafka.Demo07ProducerTest     : [run][我睡醒了]

# 在事务消息提交之后，事务消息才被 Consumer 消费到
2023-01-18 15:37:34.242  INFO 51896 --- [ntainer#6-0-C-1] c.example.kafka.consumer.Demo07Consumer  : [onMessage][线程编号:17 消息内容：Demo07Message(id=1674027443)]

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

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

什么时候能有L房呢