[poppycoderr]

Background

The event starter covered publishing: one @IntegrationEvent annotation and the event is recorded in the business transaction and delivered after commit. Consuming was left to the application. Connecting two contexts in ddk-mall took about 250 lines of consumer plumbing: creating and closing RocketMQ consumers, routing by topic and tag, decoding the payload, reading the event ID, retrying on failure, and an in-process bridge to run without a broker. Every DDK project that consumes events would write the same code.

Writing the test for it also showed that a documented guarantee was wrong, see the last section.

Changes

• IntegrationEventConsumer<T>: a consumer declares its group, its source (topic or topic:tag, the form @IntegrationEvent uses) and its payload type, and handles a ReceivedEvent<T> with the payload and the contract headers.
• IntegrationEventDispatcher routes a received message to the matching consumers and decodes the payload. With idempotent() it wraps the consumer in the inbox. It is public, so a listener for another broker can call it.
• RocketMQ consumption: one push consumer per group, started after the context is ready and stopped first on shutdown. Orderly by default (ddk.event.rocketmq.consumer.orderly).
• In-process delivery (ddk.event.local-delivery.enabled, off by default) for development and tests without a broker: after commit the event is serialized and handed to the same dispatcher on a single thread.
• Invalid declarations fail at startup: a blank group or source, or an idempotent consumer without the inbox enabled.

flowchart LR
    MQ[RocketMQ<br/>one consumer per group] --> D
    L[In-process delivery<br/>after commit, single thread] --> D
    X[Your Kafka / AMQP / JMS listener] --> D
    D[IntegrationEventDispatcher<br/>match group + source, decode JSON, read ddk-event-* headers]
    D -->|idempotent| I[IdempotentConsumer<br/>same transaction as handle]
    I --> C[IntegrationEventConsumer.handle]
    D --> C

Loading

In-process delivery is a plain listener plus a transaction synchronization, not a @TransactionalEventListener. A transactional listener for Object would make Spring Modulith write every event of the application into its publication registry.

A corrected guarantee

The README said that events with the same key keep their order. The end-to-end test published three events of one aggregate in three transactions and received them in reverse order. A key puts events into the same queue; delivery after commit runs on a thread pool, so events committed close together can be sent out of order.

• The README now has an Ordering section: what a key guarantees, how to get ordered delivery (spring.task.execution.pool.core-size=1, which the test now uses and which passed three runs in a row), and that consumers should otherwise be independent of order.
• The changelog lists the correction under Fixed.

---

背景

事件 starter 之前只管发布：加一个 @IntegrationEvent 注解，事件就会在业务事务里登记、提交后投递。消费完全留给应用自己做。在 ddk-mall 里把两个上下文连起来时，消费端的样板代码写了大约 250 行：创建和关闭 RocketMQ 消费者、按主题和标签分发、解析消息体、读取事件标识、失败重试，还有一个没有中间件时用的进程内转发。每个要消费事件的 DDK 项目都会把这些再写一遍。

为它写测试时还发现文档里的一条保证是错的，见最后一节。

改动

• IntegrationEventConsumer<T>：消费方声明消费组、来源（topic 或 topic:tag，与 @IntegrationEvent 的写法相同）和消息体类型，处理的是带消息体和契约头的 ReceivedEvent<T>。
• IntegrationEventDispatcher 把收到的消息分发给匹配的消费方并解析消息体；消费方声明 idempotent() 时用去重表包住它。它是公开的，其他中间件的监听器可以直接调用。
• RocketMQ 消费：每个消费组一个消费者，应用上下文就绪后才开始消费，关闭时最先停止。默认顺序消费（ddk.event.rocketmq.consumer.orderly）。
• 进程内转发（ddk.event.local-delivery.enabled，默认关闭）：用于没有中间件的本地开发和测试，事务提交后把事件序列化，再用单线程交给同一个分发器。
• 声明有误时启动失败：消费组或来源为空，或者要求去重却没有开启去重表。

各部分的关系见上文流程图。进程内转发用的是普通监听器加事务同步，而不是 @TransactionalEventListener：监听 Object 的事务监听器会让 Spring Modulith 把应用里的每一个事件都写进发布记录。

一条被更正的保证

README 原来说同一个 key 的事件保持顺序。端到端测试在三个事务里发布了同一个聚合的三个事件，收到时顺序是反的。key 只保证事件进入同一个队列；提交后的投递由线程池执行，相邻提交的事件可能乱序发出。

• README 新增「Ordering」一节：key 到底保证什么、怎样得到有序投递（spring.task.execution.pool.core-size=1，测试现在就是这样配置的，连续三次通过），以及否则消费方应当不依赖顺序。
• CHANGELOG 在 Fixed 下记录了这次更正。