在 Spring Boot 中发送和接收 Kafka 消息

1. 概览

Apache Kafka 是一个流行的事件流平台，用于服务之间的异步通信。Spring Boot 通过 Spring Kafka 库支持 Kafka，帮助你无需自己管理 KafkaProducer 和 poll() 循环。

在本文中，你将构建一个小示例：订单服务发送 OrderCreatedEvent 事件，consumer 接收该事件以在仓库中保留库存。读完本文后，你将了解如何配置、发送和接收 JSON 格式的消息，通过 Dead Letter Topic 处理错误，以及用 Embedded Kafka 进行测试。

2. 准备工作

2.1. Maven 依赖

你需要 spring-kafka。Spring Boot 已经管理了版本，所以无需指定版本号：

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

<!-- 用于测试 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

spring-kafka-test 提供了在内存中运行的 Kafka broker 用于测试。Awaitility 帮助你等待异步处理，无需使用 Thread.sleep()。

2.2. 本地运行 Kafka

最快的方式是使用 Apache Kafka 的官方镜像，采用 KRaft 模式（不需要 ZooKeeper）：

docker run -d --name kafka -p 9092:9092 apache/kafka:latest

broker 将在 localhost:9092 监听。

3. 数据模型和配置

要发送的事件是一个简单的 record：

package com.nonla.shop.order;

import java.math.BigDecimal;

public record OrderCreatedEvent(String orderId, String customerId, BigDecimal amount) {
}

接下来在 application.yml 中配置：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.nonla.shop.order"

各部分的含义：

• bootstrap-servers 是 broker 地址。
• Producer 对 key 使用 StringSerializer，对 value 使用 JsonSerializer。JsonSerializer 用 Jackson 将对象转换为 JSON。
• acks: all 强制 broker 只在所有副本都同步写入后才确认。这种方式更安全但稍微慢一些。
• group-id 确定 consumer group。同一 group 中的各个实例会共享 partition。
• auto-offset-reset: earliest 当 group 还没有任何 offset 提交时，consumer 从 topic 的开头读取。
• spring.json.trusted.packages 声明 JsonDeserializer 被允许创建对象的包。如果缺少这一行，你会遇到错误“The class … is not in the trusted packages”。

3.1. 自动创建 topic

Spring Boot 自动配置了一个 KafkaAdmin。这个 bean 会创建在 context 中声明的所有 NewTopic：

package com.nonla.shop.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    public static final String ORDERS_TOPIC = "orders";

    @Bean
    public NewTopic ordersTopic() {
        return TopicBuilder.name(ORDERS_TOPIC)
                .partitions(3)
                .replicas(1)
                .build();
    }
}

应用启动时，如果 topic orders 不存在，将使用 3 个 partition 创建它。如果 topic 已经存在，KafkaAdmin 会跳过而不覆盖现有配置。

4. 用 KafkaTemplate 发送消息

Spring Boot 根据 producer 配置自动创建 KafkaTemplate bean。你只需注入它：

package com.nonla.shop.order;

import com.nonla.shop.config.KafkaTopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, OrderCreatedEvent>> publish(OrderCreatedEvent event) {
        return kafkaTemplate.send(KafkaTopicConfig.ORDERS_TOPIC, event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("发送订单 {} 失败", event.orderId(), ex);
                        return;
                    }
                    var metadata = result.getRecordMetadata();
                    log.info("已发送订单 {} 至 partition {} offset {}",
                            event.orderId(), metadata.partition(), metadata.offset());
                });
    }
}

需要注意的几点：

• send(topic, key, value) 使用 orderId 作为 key。Kafka 对 key 进行哈希以选择 partition，因此同一订单的所有消息都会进入同一 partition。这样可以保证它们之间的顺序。
• 从 Spring Kafka 3.0 开始，send() 返回 CompletableFuture<SendResult> 而不是已被弃用的 ListenableFuture。
• send() 是异步的。调用立即返回，whenComplete 中的回调在 broker 响应时运行。

发送成功时，日志会类似于：

INFO  c.n.s.order.OrderEventPublisher : 已发送订单 ORD-1001 至 partition 2 offset 0

如果需要阻塞线程直到发送完成，例如在批处理任务中，你可以调用 publish(event).get(5, TimeUnit.SECONDS)。但在高流量请求处理线程中不要这样做，因为每次调用都会让一个线程等待 broker。

注意： 在同一个方法中先写入数据库再发送 Kafka 不是 一个原子事务。如果数据库 commit 成功但 Kafka 发送失败，数据将不一致。对于重要数据，你应该考虑使用 Transactional Outbox 模式。

5. 用 @KafkaListener 接收消息

consumer 端只需要一个带 @KafkaListener 注解的方法：

package com.nonla.shop.inventory;

import com.nonla.shop.config.KafkaTopicConfig;
import com.nonla.shop.order.OrderCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class OrderCreatedListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCreatedListener.class);

    private final InventoryService inventoryService;

    public OrderCreatedListener(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = KafkaTopicConfig.ORDERS_TOPIC, concurrency = "3")
    public void onOrderCreated(@Payload OrderCreatedEvent event,
                               @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                               @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("收到订单 {} 来自 partition {} offset {}", event.orderId(), partition, offset);
        inventoryService.reserve(event);
    }
}

InventoryService 只是一个普通的 service：

package com.nonla.shop.inventory;

import com.nonla.shop.order.OrderCreatedEvent;
import org.springframework.stereotype.Service;

@Service
public class InventoryService {

    public void reserve(OrderCreatedEvent event) {
        if (event.amount().signum() <= 0) {
            throw new IllegalArgumentException("金额无效：" + event.amount());
        }
        // 在仓库中保留库存...
    }
}

解释：

• topics 指定要监听的 topic。groupId 取自 spring.kafka.consumer.group-id，或者你可以直接在注解中覆盖。
• concurrency = "3" 创建 3 个 consumer 线程，每个线程处理一个 partition。值大于 partition 数量是无用的，因为多余的线程不会接收任何东西。
• @Payload 接收已反序列化的对象。@Header 获取元数据如 partition 和 offset。
• 默认情况下，当方法运行完成且没有抛出异常时 offset 被提交（AckMode.BATCH）。因此，如果应用在中途崩溃，消息将被重新处理。这是 at-least-once 语义，所以你的逻辑需要幂等：处理同一个 orderId 两次不应该保留库存两次。

如果你需要 key、timestamp 或其他任意 header，你可以接收原始的 ConsumerRecord<String, OrderCreatedEvent> 作为参数。

6. 错误处理和 Dead Letter Topic

当 listener 抛出异常时，默认的 DefaultErrorHandler 会重试 9 次，然后记录日志并跳过消息。在实际应用中，你通常希望将出错的消息转移到单独的 topic 以供后续检查：

package com.nonla.shop.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2));
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

Spring Boot 自动将这个 CommonErrorHandler bean 附加到默认的 container factory。上面的配置工作如下：

• FixedBackOff(1000L, 2) 最多重试 2 次，每次间隔 1 秒。
• 重试用尽时，DeadLetterPublishingRecoverer 将消息发送到 orders.DLT topic。这个名称是原始名称加上 .DLT 后缀，消息保留原始 partition。DLT topic 因此需要至少与原始 topic 相同数量的 partition。
• IllegalArgumentException 是数据错误，重试也无济于事。handler 会直接将消息转移到 DLT。

一条 amount = -5 的消息会产生类似的日志：

ERROR o.s.k.l.DefaultErrorHandler : Backoff none exhausted for orders-1@4
INFO  c.n.s.inventory.DltListener  : DLT 中的错误消息：ORD-1002

注意： 发生在反序列化期间的错误，例如 JSON 损坏，不会到达 listener。如果不处理，consumer 可能会永久卡在那条消息上，这称为“poison pill”。请用 ErrorHandlingDeserializer 包装反序列化器：

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

使用这个配置，反序列化错误会被传递给 DefaultErrorHandler，错误消息也会进入 DLT。

7. 用 Embedded Kafka 测试

@EmbeddedKafka 启动内存中的 broker，所以运行测试时你不需要 Docker。下面的示例测试完整的流程：通过 OrderEventPublisher 发送消息，然后验证 InventoryService 正确接收了事件。

package com.nonla.shop.order;

import com.nonla.shop.inventory.InventoryService;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.bean.override.mockito.MockitoBean;

import java.math.BigDecimal;
import java.time.Duration;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.verify;

@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(partitions = 3, topics = "orders")
class OrderKafkaIntegrationTest {

    @Autowired
    private OrderEventPublisher publisher;

    @MockitoBean
    private InventoryService inventoryService;

    @Test
    void whenOrderPublished_thenListenerReservesInventory() throws Exception {
        var event = new OrderCreatedEvent("ORD-1001", "CUS-42", new BigDecimal("250000"));

        var result = publisher.publish(event).get();
        assertThat(result.getRecordMetadata().topic()).isEqualTo("orders");

        var captor = ArgumentCaptor.forClass(OrderCreatedEvent.class);
        await().atMost(Duration.ofSeconds(10))
                .untilAsserted(() -> verify(inventoryService).reserve(captor.capture()));

        assertThat(captor.getValue()).isEqualTo(event);
    }
}

各部分的作用：

• ${spring.embedded.kafka.brokers} 是由 @EmbeddedKafka 提供的属性。它包含嵌入式 broker 的地址，我们用它来覆盖 bootstrap-servers。
• @MockitoBean（Spring Boot 3.4 中引入）用 mock 替换真实的 InventoryService，以便于验证。对于较旧的版本，使用 @MockBean。
• publish(event).get() 等待 broker 确认，这样测试可以确保消息已被写入。
• Listener 在另一个线程上运行，所以 await().untilAsserted(...) 会在最多 10 秒内重试 verify 检查。
• 由于 OrderCreatedEvent 是 record，equals() 按值比较。assertThat(...).isEqualTo(event) 因此验证 JSON 被正确序列化和反序列化。

配置中的 auto-offset-reset: earliest 帮助测试保持稳定。即使 consumer 在消息已发送之后才加入 group，它仍然可以读到该消息。

8. 常见错误和何时不应使用

你容易遇到的一些错误：

• 缺少 trusted.packages：JsonDeserializer 会拒绝创建未知的类。请正确声明包名。只在完全控制所有 producer 时才使用 "*"。
• Producer 和 consumer 在不同的包中：JsonSerializer 默认将完全限定类名写入 __TypeId__ 头。如果 consumer 服务有同名类但在不同包中，反序列化会失败。此时，在 consumer 端设置 spring.json.use.type.headers: false 和 spring.json.value.default.type: com.nonla.inventory.OrderCreatedEvent。
• Listener 运行时间长：如果一次 poll() 处理超过 max.poll.interval.ms（默认 5 分钟），broker 会认为 consumer 已死亡并重新平衡。结果是消息被重复处理。请减少 max.poll.records 或优化你的逻辑。
• concurrency 大于 partition 数量：多余的线程会闲置。要增加吞吐量，你必须先增加 partition 数量。
• 发送 null key：消息会均匀分布在各 partition 上，你会失去按实体保证顺序的能力。

Kafka 也不是在所有情况下都是好选择：

• 你需要同步的请求/响应并立即得到结果。此时使用 REST 或 gRPC。
• 系统很小，流量低，不需要保留事件历史。运行 Kafka 集群就成了浪费，RabbitMQ 或数据库中的队列表可能足够。
• 你需要复杂的按消息路由、或需要针对不同消息的延迟或优先级设置。这是传统消息 broker 的优势。

9. 总结

在本文中，你配置了 Spring Kafka 使用 JSON 序列化器并用 NewTopic 创建 topic。你通过 KafkaTemplate 以 key 发送消息来保证顺序，用带 concurrency 的 @KafkaListener 接收消息。你还用 DefaultErrorHandler 结合 Dead Letter Topic 处理错误，然后用 @EmbeddedKafka 和 Awaitility 编写了集成测试。

部署到生产环境时，请记住三点：listener 必须是幂等的，必须用 ErrorHandlingDeserializer 处理反序列化错误，发送事件必须与数据库数据保持一致。