开发者日记

外观

从 Spring Event 到 Kafka：领域事件演进避坑实录 ​

问题背景：单体里的领域事件为什么会失效 ​

在订单系统早期，“下单成功后通知库存、积分、营销服务”通常最先写成 Service 方法调用。代码能跑，但订单聚合很快开始依赖越来越多的外部能力：扣库存、发优惠券、初始化积分账户。一次下单接口可能串行调用五六个组件，任何一个非核心逻辑异常，都会拖垮主交易链路。

领域事件正是为了解决这个问题：订单聚合只负责表达“订单已创建”这一业务事实，不关心谁要响应。最常见的第一步是使用 Spring Event，它足够轻量，不需要额外中间件，适合单体应用或同进程模块解耦。

但业务增长后，Spring Event 的边界会迅速暴露：

  • 事件只存在于当前 JVM，无法被其他服务消费；
  • 默认事件监听器同步执行，异常会影响主流程；
  • @Async 可以异步执行，但缺少持久化、重试、死信和回溯能力；
  • 多实例部署后，事件实例只在发布请求所在节点处理。

因此，从 Spring Event 演进到 Kafka，不是“替换一个 API”，而是事件驱动架构从进程内解耦走向跨服务集成的过程。

方案对比：Spring Event 与 Kafka 的适用边界 ​

订单聚合发布领域事件ApplicationEventPublisher库存模块：同进程监听Kafka：跨服务广播消费者组独立消费幂等、重试、死信处理

两者不是简单的新旧替代关系：

对比项Spring EventKafka
作用范围单个 JVM多个服务、多个实例
可靠性依赖内存和线程池Broker 持久化，副本冗余
消费模型同步或异步方法调用消费者组、分区、偏移量
失败处理代码内 try/catch、自定义重试消费位点、重试队列、死信队列
适合场景同事务后扩展点、领域模型解耦跨服务集成、异步削峰、事件流

实践中更合理的判断标准是：如果监听器属于同一个部署单元，并且不需要事件留痕，用 Spring Event；如果事件需要跨服务、可靠投递或支持多个团队独立订阅，用 Kafka。

实操步骤一：先定义稳定的领域事件 ​

不要直接把数据库实体或聚合对象塞进事件。实体在事件发送后可能继续变化，还会暴露不必要字段。更稳妥的方式是定义不可变事件对象，并携带业务唯一标识。

java
package com.example.order.domain.event;

import java.time.Instant;

/**
 * 订单已创建领域事件。
 * 字段保持稳定，避免直接暴露 JPA 实体。
 */
public record OrderCreatedEvent(
        String eventId,
        Long orderId,
        Long userId,
        Long amountCents,
        Instant occurredAt
) {
}

订单聚合负责生成事件，但不直接操作 KafkaTemplate。这样领域层不依赖中间件，未来可以在本地事件、Kafka、消息云服务之间切换。

java
package com.example.order.domain.model;

import com.example.order.domain.event.OrderCreatedEvent;
import java.time.Instant;
import java.util.UUID;

public class Order {
    private Long id;
    private Long userId;
    private Long amountCents;

    public OrderCreatedEvent create() {
        // 省略参数校验、状态初始化、持久化 ID 赋值等逻辑
        return new OrderCreatedEvent(
                UUID.randomUUID().toString(),
                id,
                userId,
                amountCents,
                Instant.now()
        );
    }
}

实操步骤二：用 Spring Event 处理同进程扩展逻辑 ​

Spring 提供 ApplicationEventPublisher 发布事件。需要特别注意事务边界：如果业务方法还没提交事务，监听器就去查订单，很可能查不到数据。因此涉及数据库一致性的监听器应使用 @TransactionalEventListener(phase = AFTER_COMMIT)。

java
package com.example.order.application;

import com.example.order.domain.event.OrderCreatedEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class OrderApplicationService {

    private final ApplicationEventPublisher eventPublisher;

    public OrderApplicationService(ApplicationEventPublisher eventPublisher) {
        this.eventPublisher = eventPublisher;
    }

    @Transactional
    public void createOrder(CreateOrderCommand command) {
        // 1. 校验、计算金额、保存订单
        OrderCreatedEvent event = new OrderCreatedEvent(
                java.util.UUID.randomUUID().toString(),
                command.orderId(),
                command.userId(),
                command.amountCents(),
                java.time.Instant.now()
        );

        // 2. 同事务内只注册事件，不立即执行外部副作用
        eventPublisher.publishEvent(event);
    }
}

库存预占这类同进程逻辑可以监听该事件：

java
package com.example.order.listener;

import com.example.order.domain.event.OrderCreatedEvent;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;

@Component
public class InventoryEventListener {

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void reserve(OrderCreatedEvent event) {
        // 只有订单事务提交成功后才执行，避免脏读和空预占
        System.out.println("预占库存，订单ID=" + event.orderId());
    }
}

这一步的价值是把订单核心流程从非核心逻辑中解放出来。但要清楚：AFTER_COMMIT 只保证事务提交后执行，不保证监听器一定执行成功。应用在提交后、监听前宕机，事件仍可能丢失。

实操步骤三：演进到 Kafka 的事件发布配置 ​

当积分、营销、履约成为独立服务后，应把领域事件投递到 Kafka。下面配置使用 JSON 序列化，并开启生产者幂等，避免网络重试导致 Broker 中出现重复消息。

yaml
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
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5
    consumer:
      group-id: inventory-service
      enable-auto-commit: false
      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.example.order.domain.event"
    listener:
      ack-mode: manual_immediate

发布时建议用 orderId 作为消息 Key。相同订单会进入同一分区，既能保留单订单事件顺序，也有利于消费者按订单做局部串行处理。

java
package com.example.order.publisher;

import com.example.order.domain.event.OrderCreatedEvent;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;

@Component
public class OrderCreatedKafkaPublisher {

    private static final String TOPIC = "order-events";

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;

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

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void publish(OrderCreatedEvent event) {
        kafkaTemplate.send(TOPIC, String.valueOf(event.orderId()), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        // 记录业务事件ID，便于补偿或人工排查
                        System.err.println("事件发送失败 eventId=" + event.eventId());
                    }
                });
    }
}

这里已经解决了跨服务投递问题，但还没有彻底解决“数据库提交成功、Kafka 发送失败”的一致性缺口。高可靠场景应进一步采用 Outbox 模式：业务事务内写本地事件表，后台任务轮询未发送记录并投递 Kafka，发送成功后标记完成。

实操步骤四：消费者必须按至少一次投递设计 ​

Kafka 通常承诺至少一次投递，而不是恰好一次。消费者重启、Rebalance、业务处理成功但提交位点失败，都可能导致消息重复。因此消费逻辑必须幂等。

java
package com.example.inventory.consumer;

import com.example.order.domain.event.OrderCreatedEvent;
import java.util.Set;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class OrderCreatedConsumer {

    private static final String PROCESSED_KEY = "order-event:processed:";

    private final StringRedisTemplate redisTemplate;

    public OrderCreatedConsumer(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @KafkaListener(topics = "order-events", groupId = "inventory-service")
    public void consume(ConsumerRecord<String, OrderCreatedEvent> record,
                        Acknowledgment ack) {
        OrderCreatedEvent event = record.value();
        String processedKey = PROCESSED_KEY + event.eventId();

        // setIfAbsent 只完成快速去重，关键业务仍建议有唯一约束兜底
        Boolean firstProcess = redisTemplate.opsForValue()
                .setIfAbsent(processedKey, "1", java.time.Duration.ofDays(7));

        if (!Boolean.TRUE.equals(firstProcess)) {
            ack.acknowledge();
            return;
        }

        try {
            reserveInventory(event);
            ack.acknowledge();
        } catch (Exception ex) {
            // 删除标记，允许后续重试；如果使用唯一约束，也可依赖业务状态防重
            redisTemplate.delete(processedKey);
            throw ex;
        }
    }

    private void reserveInventory(OrderCreatedEvent event) {
        System.out.println("库存服务处理订单事件：" + event.orderId());
    }
}

生产环境不要在消费者中写无限递归重试。更推荐配置非阻塞重试和死信 Topic：短暂依赖故障进入固定或指数退避重试，参数错误、数据非法等不可恢复异常直接投递 DLT，避免阻塞分区。

避坑总结：事件驱动演进中的高频问题 ​

1. 把 Spring Event 当可靠消息队列使用 ​

常见现象是订单事务已提交，但异步监听器因线程池打满、应用重启或空指针异常没有执行，库存没有预占，且系统没有任何补偿记录。

修正方法：进程内扩展点可以使用 Spring Event，但关键业务事件要落事件表或投递 Kafka；监听器必须记录失败日志、指标，并提供补偿任务。

2. 事务提交前执行外部副作用 ​

有些团队给监听器加上 @Async 后就认为安全了，但异步线程可能在主事务提交前执行，出现“消息已发送、订单随后回滚”的问题。

修正方法：默认使用 @TransactionalEventListener(phase = AFTER_COMMIT)；需要强一致时采用 Outbox，在同一个数据库事务中写业务数据和待发送事件。

3. 消息没有业务唯一键，消费者无法幂等 ​

只依赖 Kafka offset 去重会在重复投递时失效，表现为重复扣库存、重复发券。

修正方法：每个领域事件生成全局唯一 eventId，消费侧建立唯一约束或去重表；Redis 去重只能作为加速手段，不能替代数据库最终防线。

4. 主题按“发布者想发什么”而不是“业务事实”设计 ​

一个常见坏味道是 Topic 名叫 notification，里面混着订单创建、支付成功、退款完成，所有消费者反序列化同一个大对象，字段还随发布端频繁变化。

修正方法：Topic 表达稳定业务事实，如 order-events，消息中使用 eventType 或独立事件类；新增字段保持向后兼容，避免随意修改字段类型和语义。

5. 忽略顺序与分区键的关系 ​

随机消息 Key 或用用户 ID、事件 ID 作为 Key，可能导致同一订单的创建、支付、取消事件进入不同分区，被不同消费者并发处理，出现状态乱序。

修正方法：同一聚合根使用相同分区键，订单生命周期事件统一使用 orderId；消费者侧还要根据事件发生时间或聚合版本做状态校验。

结语 ​

事件驱动架构的核心不是引入消息中间件，而是让系统围绕稳定的业务事实协作。Spring Event 适合在单体内部保持领域模型干净，Kafka 适合承载跨服务、可持久化、可独立消费的领域事件。真正稳妥的演进路径通常是：先抽象领域事件，再规范事务后发布，随后迁移 Kafka，最后补齐 Outbox、幂等、重试、死信和可观测能力。这样既不会在早期被复杂基础设施拖慢，也不会在业务增长后被同步调用和消息丢失反复惩罚。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。

本文来自开发者日记（devdiary.cn）· 记录开发者实战笔记与免费在线工具箱

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。