Spring Boot Kafka消费者异步性及日志格式配置问题咨询

背景信息 #

我们使用Spring Boot 2.7集成Kafka，消费消息的核心代码如下：

@KafkaListener(topics = "${kafka.topic.stuff}")
public void consume(@Payload String message) {
    log.info("in kafka consumer");
    // process event
}

不确定该消费逻辑是否为同步处理（即单条消息处理过慢会阻塞后续消息）。原本打算通过日志线程名判断，但Kafka相关日志未按自定义格式输出线程信息，日志示例如下：

2025-01-06T17:40:00,143Z INFO  pool-2-thread-1 c.c.g.scheduler.SomemScheduler [correlationToken:ANP-06c003a1-d31c-42f6-9d2d-a9cdb5bfde96] => some event scheduler started..

2025-01-06T17:40:07,721Z INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 c.c.g.s.cdm.consumer.MyConsumer [] => in kafka consumer
2025-01-06T17:40:14,665Z INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 org.apache.kafka.clients.NetworkClient [] => [Consumer clientId=consumer-SWH-1, groupId=SWH] Node -1 disconnected.

自定义logback.xml的格式配置如下：

<pattern>%date{"yyyy-MM-dd'T'HH:mm:ss,SSSXXX", UTC} %-5level %thread %logger{42} [%X{correlationTokenKV}] => %msg%n</pattern>

依赖引入方式：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

问题与解答 #

1. Kafka消费者是否为异步处理？ #

默认情况下，Spring Kafka的@KafkaListener是同步处理的：

• 单个消费者线程会按顺序拉取消息，逐条执行consume方法，处理完一条才会处理下一条。
• 如果单条消息处理耗时过长，会阻塞后续消息的消费，直到当前消息处理完成（包括提交偏移量）。
• 从日志也能看出，消费消息和Kafka客户端日志共用同一个线程org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1，说明是单线程同步执行。

2. 若不是，如何实现异步消费？ #

有三种常见实现方式：

• 方式一：消费方法内异步执行业务逻辑
  把耗时业务逻辑丢到线程池处理，快速释放消费者线程：

  @Autowired
  private ThreadPoolTaskExecutor taskExecutor;
  
  @KafkaListener(topics = "${kafka.topic.stuff}")
  public void consume(@Payload String message) {
      log.info("in kafka consumer");
      taskExecutor.execute(() -> {
          // 处理耗时业务逻辑
      });
  }
  

  注意：这种方式需自行处理异常，且消息偏移量会在consume方法执行完成后立即提交，若异步逻辑失败可能丢失消息。

• 方式二：使用Spring Kafka异步消费特性
  将消费方法定义为返回CompletableFuture，框架会自动异步处理，偏移量在Future完成后提交：

  @Autowired
  private ThreadPoolTaskExecutor taskExecutor;
  
  @KafkaListener(topics = "${kafka.topic.stuff}")
  public CompletableFuture<Void> consume(@Payload String message) {
      log.info("in kafka consumer");
      return CompletableFuture.runAsync(() -> {
          // 处理业务逻辑
      }, taskExecutor);
  }
  

• 方式三：增加消费者并发数
  通过配置参数设置多个消费者线程，并行消费不同分区的消息：

  spring.kafka.listener.concurrency=3
  

  注意：并发数不能超过Topic的分区数，否则多余线程会空闲。

3. 为何自定义标准日志格式未生效？ #

核心原因是Kafka客户端（org.apache.kafka包下类）的日志配置被单独覆盖：

• Spring Boot默认会为Kafka客户端日志配置单独的Appender或格式，若你的logback.xml仅配置了根日志，未对Kafka相关包做明确配置，就会导致自定义格式不生效。
• 从日志示例看，自定义的correlationTokenKV在Kafka相关日志中为空，且线程名位置显示容器名称，说明Kafka客户端日志未使用你定义的格式。

4. 如何让Kafka使用自定义日志格式或至少输出线程名？ #

需在logback.xml中为Kafka相关包明确指定使用自定义格式，示例配置如下：

<!-- 自定义控制台Appender，包含你的格式 -->
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
    <encoder>
        <pattern>%date{"yyyy-MM-dd'T'HH:mm:ss,SSSXXX", UTC} %-5level %thread %logger{42} [%X{correlationTokenKV}] => %msg%n</pattern>
    </encoder>
</appender>

<!-- 为Kafka相关包绑定自定义Appender -->
<logger name="org.apache.kafka" level="INFO" additivity="false">
    <appender-ref ref="CONSOLE"/>
</logger>

<logger name="org.springframework.kafka" level="INFO" additivity="false">
    <appender-ref ref="CONSOLE"/>
</logger>

<!-- 根日志配置 -->
<root level="INFO">
    <appender-ref ref="CONSOLE"/>
</root>

配置后，Kafka相关日志会使用自定义格式，正确输出线程名和correlationTokenKV信息。

内容的提问来源于stack exchange，提问作者John Little