DDK Event Starter

发布聚合根登记的领域事件；标了 @IntegrationEvent 的事件交给 Spring Modulith，在业务事务里登记、提交后可靠地投递到进程外；消费端用 IdempotentConsumer 去重。

text

aggregate.registerEvent(e) ──► 仓储写入 ──► DomainEventPublisher ──► Spring ApplicationEventPublisher
                                                                        │
          进程内   @EventListener / @TransactionalEventListener ◄───────┤
          进程外   @IntegrationEvent + Spring Modulith     ◄───────────┘
                   在业务事务中写入 event_publication，提交后投递到 Kafka / AMQP / JMS / MessageChannel

为什么复用 Spring Modulith

事务性 Outbox 的发布侧，Spring Modulith 已经做好了：事件发布记录在业务事务里写入，提交后投递，失败的记录停在 FAILED 可以重新提交，还有过期监控和重启重投。验证确认它的 JDBC 实现与 MyBatis-Plus 共用同一个 DataSource 和事务：提交后有记录，回滚后没有。DDK 按「不重复实现 Spring Modulith 已有能力」的原则直接复用，只补两处缺口：

领域层怎样声明对外事件。Modulith 用自己的 @Externalized 或 jMolecules 注解选择事件，领域层不能依赖它们。DDK 在 com.ddk.core.domain 提供纯 JDK 的 @IntegrationEvent，由本 starter 翻译给 Modulith。
消费端去重。投递是「至少一次」，Modulith 只管发布侧。

集成事件

java

@IntegrationEvent(value = "user-events", key = "userId", id = "eventId")
public record UserRegisteredEvent(UUID eventId, UserId userId, String username, Instant occurredOn) implements DomainEvent {
}

属性	含义
`value`	投递目标：Kafka topic、AMQP exchange、JMS 目的地或 `MessageChannel` Bean 名，取决于引入的外发模块
`key`	作为消息 key 的访问方法名。类型化标识取原始值；同一个 key 的消息进入同一个分区，它保证什么、不保证什么见顺序
`id`	事件自带标识的访问方法名，取值写进消息头 `ddk-event-id`。必须是事件自身的数据，重投前后才保持不变
`type` / `version`	契约名称与版本，见下一节

应用再引入 Modulith 的 JDBC 发布记录和一个外发模块，版本由 DDK 的 BOM 管理：

xml

<dependency>
    <groupId>org.springframework.modulith</groupId>
    <artifactId>spring-modulith-starter-jdbc</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.modulith</groupId>
    <artifactId>spring-modulith-events-kafka</artifactId>
</dependency>

消息体里的类型化标识写成原始值（"userId":42），由本 starter 注册的 IdentifierJacksonModule 处理。应用自己声明了 EventExternalizationConfiguration 时，DDK 的选择规则让位。

必须在事务里发布

集成事件靠事务送出去：发布时登记，提交之后投递。发布时外面没有事务，事件就不会投递；用了 Spring Modulith 时还会留下一条一直处于未完成状态的发布记录，日后重新投递未完成的事件时可能被发出去。

0.7.0 起，DDK 在发布的那一刻检查，默认直接抛出异常：

text

IllegalStateException: com.acme.order.OrderPaid was published outside a transaction and will not be delivered:
publish integration events inside the transaction that changes the data, for example from a @Transactional service method

检查发生在 Modulith 登记事件之前，所以报错时表里不会留下记录。ddk.event.outside-transaction=warn 改成只记警告，ignore 保持沉默；这两种方式下未完成的记录照旧会写入。0.6.0 及更早的版本没有这项检查，事件会悄悄丢失。

投递到 RocketMQ

Spring Modulith 没有 RocketMQ 模块，由本 starter 补上。引入 RocketMQ 客户端（版本由 DDK 的 BOM 管理）代替 Modulith 的外发模块，再配置 NameServer：

xml

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
</dependency>

yaml

ddk:
  event:
    rocketmq:
      name-server: localhost:9876

mermaid

sequenceDiagram
    participant App as 应用服务
    participant DB as event_publication
    participant L as DDK RocketMQ 监听器
    participant MQ as RocketMQ
    App->>DB: 在业务事务里登记事件
    Note over App,DB: 提交
    DB-->>L: 提交之后
    L->>MQ: 发到 topic:tag，按 key 哈希选队列
    MQ-->>L: SEND_OK
    L->>DB: 标记发布完成

方面	行为
目标	`value` 写成 `topic` 或 `topic:tag`，与 rocketmq-spring 的约定一致
key	有 key 的消息按 key 哈希选队列，同一个聚合的事件进入同一个队列（见顺序）；key 同时写进消息的 keys，方便在控制台按业务标识查消息
消息头	`ddk-event-type`、`ddk-event-version`、`ddk-event-id` 写成用户属性，消费方用 `MessageExt.getUserProperty(...)` 读取
消息体	用应用的 `JsonMapper` 序列化成 JSON；`String` 和 `byte[]` 原样发送
失败	发送结果不是 `SEND_OK` 就算失败，发布记录保持未完成，可以重投
producer	应用已有 `DefaultMQProducer`（例如 rocketmq-spring 注册的那个）时直接用它，否则按 `ddk.event.rocketmq.*` 创建

它工作在 Modulith 默认的监听器模式下；spring.modulith.events.externalization.enabled=false 或 mode=outbox 时不注册。测试用 Testcontainers 起真实的 RocketMQ broker。

顺序

key 让同一个聚合的事件进入同一个分区或队列，中间件保持它们到达的先后。它不保证事件按提交的先后到达。

提交后的投递由 Spring 的任务执行器执行，默认是线程池。相邻提交的两个事件由两个线程发出，后提交的可能先到。DDK 的端到端测试里，同一个聚合在三个事务里先后发出的三个事件，到达消费方时顺序是反的。

需要	做法
同一个聚合的事件按提交顺序到达	设置 `spring.task.execution.pool.core-size=1`，投递按提交的先后一个一个发出。代价是投递吞吐受限；开启虚拟线程后每次投递各用一个线程，这个办法不成立
吞吐	保留线程池，让消费方不依赖顺序：事件里带上聚合版本或足够的状态，晚到的事件忽略或补偿

失败记录重投同样会打乱顺序，所以依赖顺序的消费方无论如何都是脆弱的。限界上下文集成一文里有按聚合版本判断的做法；ddk-mall 示例的做法是补偿：订单已取消之后才收到「库存已预占」，就再发一次「订单已取消」。

消费事件

把消费方声明成 Bean，订阅、解析消息体、读取契约头和去重都由 DDK 完成：

java

@Component
public class OrderPlacedConsumer implements IntegrationEventConsumer<OrderPlacedPayload> {

    @Override
    public String group() {
        return "inventory";
    }

    @Override
    public String source() {
        return "order-events:placed";   // 写法与 @IntegrationEvent(value) 相同：topic 或 topic:tag
    }

    @Override
    public Class<OrderPlacedPayload> payloadType() {
        return OrderPlacedPayload.class;
    }

    @Override
    public void handle(ReceivedEvent<OrderPlacedPayload> event) {
        inventoryService.reserve(event.eventId(), event.payload());
    }
}

消费方和控制器一样是用例的入口：放在适配层，只调用应用服务。
消息体类型由消费方自己定义，是发布方契约在消费方这边的一份拷贝，只声明用得到的字段。消费方因此不依赖发布方的类。
ReceivedEvent 带着 ddk-event-* 消息头里的 eventId、type、version。contract() 声明消费方认识的类型和版本，见消费方契约。
投递是至少一次：handle 抛异常，消息会被重新投递。
idempotent() 返回 true 时，DDK 用 IdempotentConsumer 按事件标识包住 handle。需要开启 ddk.event.inbox.enabled，并且发布方声明了事件标识；否则启动时或收到第一条消息时报错并说明原因。
去重表按 <group>:<consumerId()> 记录一条事件已被处理。consumerId() 默认是实现类的简单类名，取的是用户类而不是 Spring 代理。同一个 group 里两个幂等消费方身份相同时会互相跳过对方的事件，所以启动直接失败：同名的两个类、同一个类的两个 Bean、匿名类，都要显式覆盖 consumerId()。改类名会改变默认身份，已处理过的事件被重投时会再处理一次；改名前先把 consumerId() 固定为旧名字。被代理的幂等消费方（例如 handle 上有 @Transactional）在 0.9 之前记录的是代理类名，升级后重投的旧事件会再处理一次，介意时临时把 consumerId() 设为旧名字（Name$$SpringCGLIB$$0）。

消息从哪里来：

来源	启用条件	行为
RocketMQ	classpath 上有 `rocketmq-client`，并配置了 `ddk.event.rocketmq.name-server`	每个消费组一个消费者，订阅这个组的全部主题和标签。默认顺序消费：处理失败的消息会暂停它所在的队列并重试，后面的消息不会越过它
进程内	`ddk.event.local-delivery.enabled=true`	用于没有中间件的本地开发和测试。事务提交后把事件序列化，用单线程交给同一个应用里的消费方，走同样的解析过程。不持久，失败只记日志
其他	自己的监听器	在 Kafka、AMQP、JMS 的监听器里调用 `IntegrationEventDispatcher.dispatch(group, topic, tag, headers, json)`，复用解析、消息头和去重

加锁的消费方不要用 idempotent()。 去重表会在进入 handle 之前开启事务，锁就落在了事务里面，释放锁时事务还没提交。这种情况自己调用 IdempotentConsumer，把它放在锁的里面：它在没有事务时会自己开启一个，所以 aggregateLocks.executeAll("sku", skuIds, () -> idempotentConsumer.handle(consumer, eventId, () -> ...)) 一条语句就是「锁 → 事务 → 去重登记」。顺序写反时，Concurrency Starter 会在加锁处直接报错（0.7.0 起）。

链路与指标

0.6.0 起，发布和消费都是 Micrometer 的观测（Observation）。应用里有什么就用什么，没有 ObservationRegistry 时什么都不做。

观测	时机	标签
`ddk.event.publish`	事件发往 RocketMQ，或交给进程内转发	`messaging.destination.name`
`ddk.event.consume`	一条消息被一个消费组的消费方处理	`messaging.destination.name`、`messaging.consumer.group.name`

应用里有链路追踪时，发布方把链路上下文写进消息头，消费方从消息头接上。一次请求经过几个服务，或者一个应用里的几个上下文，都是同一条链路。

两点要知道：

事件是提交之后在线程池里发出的，链路上下文要被带到那个线程上才接得起来。DDK Tracer Starter 会配好这件事；不用它时自己声明一个 ContextPropagatingTaskDecorator Bean。
重新投递的事件会开启新链路。 投递失败或重启之后再次投递时，原来的链路上下文已经不在了：它没有随事件发布记录保存。

接入其他消息中间件时，监听器带着消息头调用 IntegrationEventDispatcher.dispatch，消费端的观测和链路接续就自动生效。

事件契约

每条投递出去的集成事件都带契约消息头：

消息头	来源	用途
`ddk-event-type`	`type()`，默认取类的简单名	消费方按这个稳定名称分发，而不是按 Java 类；移动包、改类名不影响下游。改类名时把旧名称写进 `type()`
`ddk-event-version`	`version()`，默认 `1`	消息体是哪个版本的结构
`ddk-event-id`	`id()` 指向的访问方法	去重，见消费端幂等
`ddk-event-tenant`	`tenant()` 指向的访问方法；没有声明或值为 `null` 时不写	事件所属的租户，见多租户

演进规则：

text

同一版本内     只新增可选字段；消费方必须忽略不认识的字段
破坏性修改     删除、改名字段，或改变字段的类型、含义
              → 新建一个 type() 相同、version = N + 1 的事件类
              → 两个版本并行发布，直到所有消费方都切换过去；
                消费方在 contract() 里声明自己认识的版本
              → 再删除旧版本

消费方契约

消费方不覆盖 contract() 时，主题和标签匹配的消息都交给它，不看类型和版本。破坏性变更期间两个版本发到同一个地方，只认识版本 1 的消费方会把版本 2 的消息解析成版本 1 的消息体，变化的部分被悄悄忽略。声明契约可以避免这种情况：

java

@Override
public EventContract contract() {
    return EventContract.of("order.placed", 1);   // 这个消费方认识的类型和版本
}

消息	不声明契约	`EventContract.of(type, versions...)`
类型不同，或者没有类型头	处理	不交给这个消费方
类型相同，版本在列表里	处理	处理
类型相同，版本不在列表里、没有版本头、版本头不是整数	处理；缺失或非法时 `version` 为 0	跳过并记警告日志，不解析消息体

默认跳过，是因为并行发布需要这样：旧消费方继续处理版本 1，忽略版本 2 的那份。不打算并行发布、希望遇到未知版本就停下来等升级的，加上 .rejectOtherVersions()：这条消息处理失败，由消息中间件重投。顺序消费（默认）时同一队列后面的消息会等着；并发消费时按 RocketMQ 的重试上限，之后进入死信队列。

类型为空、没有版本、版本小于 1，都会在启动时失败。参考应用 ddk-mall 的 8 个消费方都声明了契约。

启动时，本 starter 扫描应用包里所有 @IntegrationEvent，声明有误就直接启动失败：key()、id() 或 tenant() 不是无参访问方法、版本号小于 1、目标为空、两个类声明了相同的类型和版本。否则这些错误要到事务提交后的投递环节才暴露，那时发布记录已经停在 FAILED。

多租户

OperatorContext 是线程上的。事件在提交之后由另一个线程发送，失败的还可能在重启之后重发，发布方的上下文带不过去。租户必须和事件标识一样，是事件自己的数据：

java

@IntegrationEvent(value = "order-events:placed", id = "eventId", tenant = "tenantId")
public record OrderPlacedMessage(String eventId, Long tenantId, Long orderId) {
}

它的值写进消息头 ddk-event-tenant。消费方选择是否在这个租户里处理：

java

@Override
public boolean runInEventTenant() {
    return true;
}

选择之后，handle 和去重登记都以系统操作者 ddk-event（IntegrationEventDispatcher.OPERATOR_ID）加事件租户的身份执行，用的是 OperatorContext.runAs，结束后恢复监听线程原来的上下文。
消息没有这个头时处理失败，由消息中间件重投，不会在没有租户的情况下执行。
没有选择的消费方不读取这个头，照旧在没有操作者的上下文里执行。消息里的租户是发布方的声明，是否据此切换身份由消费方针对它信任的来源决定。
事件标识应当跨租户唯一，例如 UUID：去重表只按消费方和事件标识记录。

消费端幂等

java

@KafkaListener(topics = "user-events")
void on(UserRegisteredEvent event, @Header("ddk-event-id") String eventId) {
    idempotentConsumer.handle("welcome-mail", eventId, () -> mailService.sendWelcome(event.userId()));
}

用 RocketMQ 时，事件 ID 从用户属性里取：message.getUserProperty("ddk-event-id")。

text

handle(consumer, messageId, handler)            加入调用方的事务，没有则新开
  保存点：INSERT (consumer, message_id)          主键冲突 → 回滚到保存点，返回 false
  handler.run()                                  抛异常 → 整个事务连同登记一起回滚
  return true

登记与处理逻辑的写入在同一个事务里：处理失败时消息保持未处理，重投后会再次执行。
登记放在保存点之后：PostgreSQL 在语句失败后会中止整个事务，重复消息只回滚到保存点，调用方的其他写入照常提交。这一点有基于真实 PostgreSQL 的测试。
purgeOlderThan(Duration) 清理旧记录，保留时长要超过中间件可能重投的最长时间。定时维护可以按计划做这件事。

开启 ddk.event.inbox.enabled=true 后，启动时用 CREATE TABLE IF NOT EXISTS 创建 ddk_processed_message，适用于 MySQL、PostgreSQL、H2；其他数据库或由迁移工具管理表结构时，关闭 initialize-schema 并自行建表。

运维

投递失败的发布留在 Spring Modulith 的 event_publication 表里；默认的完成模式下，已完成的也留在那里。去重表保存每一条处理过的消息。这些都不会自己清理。开启 ddk.event.maintenance.enabled=true 后，一个后台线程通过 Modulith 和去重表自己的接口来做，从不直接改表：

text

每隔 ddk.event.maintenance.interval（1m）
  FailedEventPublications.resubmit(minAge = resubmit-min-age, batchSize = resubmit-batch-size)
  CompletedEventPublications.deletePublicationsOlderThan(delete-completed-after)     为 0 时保留
  IdempotentConsumer.purgeOlderThan(inbox-retention)                               为 0 时保留

多个实例。 Modulith 重投时不认领发布记录（JDBC 仓储没有覆盖 markResubmitted，默认实现直接返回 true），两个实例同时重投会把同一条发两次。应用里有 ShedLock 的 LockProvider 时，每一轮在名为 ddk-event-maintenance 的锁里执行，同一时刻只有一个实例在维护；ddk-job-starter 就提供一个。没有 LockProvider 时启动会打一条警告：只在一个实例上开启，或者依靠消费端去重。
卡住的发布。 只有 Modulith 标记为 FAILED 的发布会被重投。停在 PUBLISHED 或 PROCESSING 的（例如发送到一半进程挂了），要开启 Modulith 的陈旧检测才会被标记为失败，例如 spring.modulith.events.staleness.processing=5m、spring.modulith.events.staleness.published=5m。
重放窗口。 重投和中间件的重新投递都会让一条消息再来一次。inbox-retention 要长于这件事可能发生的最长时间，否则记录被清掉之后，去重消费方会把它当成新消息再处理一次。默认 7 天。
每一步单独失败：一步出错只记日志，其他步骤和下一轮照常进行。

有 Micrometer 时，ddk.event.publications{status} 给出 published、processing、failed、resubmitted 各有多少条，不统计已完成的，在采集时查询。failed 长时间大于零，或者 published 持续增长时告警。

配置项

配置项	默认值	说明
`ddk.event.enabled`	`true`	注册基于 Spring 的 `DomainEventPublisher`
`ddk.event.inbox.enabled`	`false`	注册 `IdempotentConsumer`
`ddk.event.inbox.table`	`ddk_processed_message`	已处理消息表
`ddk.event.inbox.initialize-schema`	`true`	启动时建表
`ddk.event.rocketmq.name-server`		NameServer 地址，多个用 `;` 分隔；设置后由 DDK 创建 producer
`ddk.event.rocketmq.producer-group`	`ddk-event-producer`	该 producer 的生产者组
`ddk.event.rocketmq.send-timeout`	`3s`	单次发送超时
`ddk.event.rocketmq.consumer.enabled`	`true`	应用声明了消费方时，是否从 RocketMQ 消费
`ddk.event.rocketmq.consumer.orderly`	`true`	顺序消费；关闭后并发消费，失败的消息单独重投
`ddk.event.outside-transaction`	`fail`	集成事件在事务之外发布时：`fail` 在发布处抛出异常，`warn` 记警告，`ignore` 不处理（0.7.0 起）
`ddk.event.local-delivery.enabled`	`false`	在进程内把集成事件送给本应用里的消费方，仅用于开发和测试
`ddk.event.maintenance.enabled`	`false`	开启上面「运维」里的定时维护
`ddk.event.maintenance.interval`	`1m`	两轮维护之间的间隔
`ddk.event.maintenance.resubmit-failed`	`true`	是否重投失败的发布
`ddk.event.maintenance.resubmit-min-age`	`1m`	只重投失败了至少这么久的
`ddk.event.maintenance.resubmit-batch-size`	`100`	每轮最多重投多少条
`ddk.event.maintenance.delete-completed-after`	`7d`	删除早于这个时间的已完成发布；`0` 表示保留
`ddk.event.maintenance.inbox-retention`	`7d`	删除早于这个时间的已处理消息记录；`0` 表示保留。要长于任何重投、重新投递的时间窗口
`spring.modulith.events.*`	Spring Modulith	发布记录表、重启重投、完成模式、过期监控

不适用的场景 / 已知问题

乱序：投递顺序没有保证，原因和对策见顺序一节。
死信处理：失败的发布记录停在 FAILED，需要自己接告警，并通过 Modulith 的 FailedEventPublications 重新提交。

DDK Event Starter ​

为什么复用 Spring Modulith ​

集成事件 ​

必须在事务里发布 ​

投递到 RocketMQ ​

顺序 ​

消费事件 ​

链路与指标 ​

事件契约 ​

消费方契约 ​

多租户 ​

消费端幂等 ​

运维 ​

配置项 ​

不适用的场景 / 已知问题 ​

DDK Event Starter

为什么复用 Spring Modulith

集成事件

必须在事务里发布

投递到 RocketMQ

顺序

消费事件

链路与指标

事件契约

消费方契约

多租户

消费端幂等

运维

配置项

不适用的场景 / 已知问题