easy-event-bus 事件总线组件设计

本文档描述 easy-event-bus 组件的设计与实现。该组件提供领域事件发布/订阅完整链路，底层采用 SPI 分层架构：业务门面 → MessageProducer/Consumer 装饰器层 → MessageProducerImplementation/ConsumerImplementation SPI → easy-messaging Adapter → 实际 MQ 后端（RocketMQ / Redis / JDBC / ZeroMQ / Kafka，可选）。组件约 60% 为 Spring-free 纯 POJO，可在无容器环境下单独测试。

更名记录：组件由 easy-cqrs 更名为 easy-event-bus。框架采用 CQS 范式，本组件专注领域事件总线；原命令路由（commands 包）从未正式启用且不再发展，仅作为遗留代码保留。Java 包名 io.matrix.cqrs 与类名（CqrsMessagingAutoConfiguration 等）保持不变，配置前缀不受影响。

• 一、背景与问题
• 二、组件定位与边界
• 三、总体架构
• 四、核心 API 草图
• 五、SPI 分层架构
• 六、包结构与核心类
• 七、消息装饰链
• 八、配置与后端
• 九、使用示例
• 十、关键设计决策
• 十一、测试覆盖

一、背景与问题

1.1 为什么要独立事件总线层

Matrix Framework 已有 easy-messaging 提供 MQ 抽象，但业务层每次做事件通信都要：

1.2 与 Axon / EventBus 的区别

二、组件定位与边界

定位：后端中立的领域事件门面。业务声明 BaseDomainEvent 后，通过 DomainEventPublisher 发布，通过 DomainEventDispatcher 消费。后端 MQ 可替换，消费端支持幂等 + 拦截器链。

边界：

约束：

• MessageProducerImplementation / MessageConsumerImplementation 是 SPI 契约，新后端只要实现这两个接口即可接入
• DomainEventDispatcher.initialize() 标 @PostConstruct——纯单元测试需手动调用
• hutool ClassUtil.loadClass 硬依赖——事件反序列化阶段用它加载事件类（fallback 到 handler.getEventClass()）
• 所有 MQ 后端均为 optional scope——编译时不强制，运行时按需引入

三、总体架构

flowchart TD
    subgraph "业务层"
        A2["BaseDomainEvent (kernel-client 提供)"]
        A3["业务 Service：publish(event)"]
    end

    subgraph "Facade 层"
        B2["DomainEventPublisherImpl<br/>DomainEventPublisher"]
        B3["DomainEventDispatcher<br/>initialize() + subscribe + handleMessage"]
    end

    subgraph "Decorator 层"
        C1["MessageProducerImpl<br/>prepareHeaders + pre/post Interceptor + send"]
        C2["MessageConsumerImpl<br/>ChannelMapping + DecoratedMessageHandlerFactory"]
        C3["MessageHandlerDecoratorChainBuilder<br/>递归组装装饰链"]
        C4["DecoratedMessageHandlerFactory<br/>按 order 排序 → Chain"]
    end

    subgraph "SPI 层 (后端中立)"
        D1["MessageProducerImplementation<br/>接口：send() + setMessageIdIfNecessary + withContext"]
        D2["MessageConsumerImplementation<br/>接口：subscribe + groupByChannel + getId + close"]
    end

    subgraph "Adapter 层 (easy-messaging 桥接)"
        E1["EasyMessagingProducerAdapter<br/>ProducerImplementation → BackendProducer"]
        E2["EasyMessagingConsumerAdapter<br/>ConsumerImplementation → BackendConsumer"]
    end

    subgraph "easy-messaging (compile scope 依赖)"
        F1["BackendProducer"]
        F2["BackendConsumer"]
    end

    subgraph "MQ 后端 (optional)"
        G1["RocketMQ"]
        G2["Redis"]
        G3["JDBC"]
        G4["ZeroMQ"]
        G5["Kafka"]
    end

    A2 --> B2 --> C1 --> D1
    C1 --> E1 --> F1
    F1 --> G1
    A3 --> B3 --> C2 --> D2
    C2 --> E2 --> F2
    F2 --> G2
    B3 -- "@PostConstruct initialize()" --> C3 --> C4

关键设计：业务层只与 Facade 层（DomainEventPublisher / DomainEventDispatcher）交互；后端切换只需换 SPI 实现 + Adapter，业务零感知。

四、核心 API 草图

4.1 消息抽象

public interface Message {
    String ID = "ID";
    String PARTITION_ID = "PARTITION_ID";
    String DESTINATION = "DESTINATION";
    String DATE = "DATE";

    String getId();
    String getPayload();
    Map<String, String> getHeaders();
    String getHeader(String name);
    String getRequiredHeader(String name);   // 缺失抛 RuntimeException
    boolean hasHeader(String name);
    // ...
}

// MessageBuilder（链式 + 前缀 extraHeaders）
Message m = MessageBuilder.withPayload(body)
    .withHeader("k", "v")
    .withExtraHeaders("x-", extras)        // 自动拼前缀
    .build();

4.2 Producer SPI + Facade

// SPI：后端实现
public interface MessageProducerImplementation {
    void send(Message message);
    default void setMessageIdIfNecessary(Message message) {}  // 默认 no-op
    default void withContext(Runnable runnable) { runnable.run(); }  // 默认直接跑
}

// Facade 装饰器：prepareHeaders + pre/post 拦截器 + send
public final class MessageProducerImpl implements MessageProducer {
    // prepareHeaders: setMessageIdIfNecessary → DESTINATION (ChannelMapping) → PARTITION_ID (UUID if absent)
    // preSend → implementation.send(message) → postSend(null) / postSend(exception)
    // implementation.withContext(() -> send(message))
}

4.3 Event 发布/订阅

// 发布（Facade）
DomainEventPublisher publisher = ...;
publisher.publish(aggregateType, aggregateId, events, headers);
// 内部：序列化 BaseDomainEvent → Message（带 AGGREGATE_ID / AGGREGATE_TYPE / EVENT_TYPE）
// → MessageProducer.send(channel, message)

// 订阅（构建）
DomainEventHandlers handlers = DomainEventHandlersBuilder
    .forAggregateType("com.example.Order")
    .onEvent(OrderPaidEvent.class, envelope -> { ... })
    .onEvent(OrderShippedEvent.class, envelope -> { ... })
    .andForAggregateType("com.example.Invoice")
    .onEvent(InvoiceCreatedEvent.class, envelope -> { ... })
    .build();

// 初始化（@PostConstruct）
DomainEventDispatcher dispatcher = new DomainEventDispatcher("dispatcher-1", handlers, messageConsumer, nameMapping);
dispatcher.initialize();  // 生产由 Spring @PostConstruct 触发；纯单测需手动调

4.4 DomainEventDispatcher 处理流程

flowchart TD
    A["MessageConsumer 收到 Message"]
    B["handleMessage(msg, handlers)"]
    C["externalEventTypeToEventClassName<br/>把外部 topic 事件名转回 Class.getName()"]
    D{"遍历 handlers<br/>handler.accept(message) 匹配？"}
    E["构造 handler 级幂等 key<br/>dispatcherId:aggregateType:eventName"]
    F["detector.doWithMessage<br/>占位即标记（重复则跳过）"]
    G["handle：ClassUtil.loadClass + JsonMapper.fromJson<br/>handler.invoke(envelope)"]
    H["warn 日志记录 handlerKey 与 messageId<br/>继续后续 handler（异常隔离不重投）"]
    I["处理下一个 handler"]
    Z["结束（无匹配 handler）"]

    A --> B --> C --> D
    D -- "匹配" --> E --> F
    F -- "首次消费" --> G
    F -- "重复消息" --> I
    G --> I
    G -- "抛异常" --> H --> I
    D -- "不匹配" --> I
    I --> D

消费语义（v1.1.0 起）：

• 异常隔离：单个 handler 异常只记 warn 日志，不影响同一消息的其他 handler，消息不重投
• handler 级幂等：幂等判定下沉到每个 handler 粒度（详见 §7.1）

五、SPI 分层架构

业务代码
  ↓
DomainEventPublisher / DomainEventDispatcher
  ↓
MessageProducerImpl / MessageConsumerImpl     ← Decorator 层（Spring-free，可测）
  ↓                                                          ↓
MessageProducerImplementation                MessageConsumerImplementation  ← SPI 接口（Spring-free，可测）
  ↓                                                          ↓
EasyMessagingProducerAdapter                 EasyMessagingConsumerAdapter    ← Adapter 层（桥接 easy-messaging）
  ↓                                                          ↓
BackendProducer / BackendConsumer            ← easy-messaging（compile scope）
  ↓
RocketMQ / Redis / JDBC / ZeroMQ / Kafka      ← MQ 后端（optional scope）

为什么分层？

1. Decorator 层与 SPI 层：MessageProducerImpl.send() 包含 prepareHeaders / ChannelMapping / PartitionId 自动生成 / 拦截器 pre/post。这些逻辑和具体 MQ 实现无关，放在 Decorator 层让后端实现只关心”怎么把 Message 发出去”。
2. Adapter 层：easy-messaging 提供 BackendProducer/SendMessage API，Adapter 把 MessageProducerImplementation 转成 BackendProducer。业务层看不到 Adapter 的存在——CqrsMessagingAutoConfiguration（类名为历史命名，保持不变）在启动时自动装配。
3. Optional scope 后端：MQ 后端（RocketMQ / Redis / JDBC / ZeroMQ / Kafka）全部 optional scope，编译时不强制，运行时按需引入 + easy.messaging.backend=rocketmq 选后端。

六、包结构与核心类

easy-event-bus/
└── io/matrix/cqrs/                   # 包名保持不变（历史命名）
    ├── core/                          # ✅ Spring-free，纯 POJO/SPI，可单独测
    │   ├── Message.java                  # 接口：ID/PARTITION_ID/DESTINATION/DATE
    │   ├── MessageImpl.java              # 实现：payload + headers Map
    │   ├── MessageInterceptor.java       # 接口：preSend/postSend + preReceive/postReceive
    │   ├── ChannelMapping.java           # 接口：transform(channel) → String
    │   ├── DefaultChannelMapping.java    # Builder 实现
    │   ├── SubscriberIdAndMessage.java   # 组合：subscriberId + Message
    │   ├── producer/
    │   │   ├── MessageBuilder.java       # 链式构建：withPayload / withHeader / withExtraHeaders / build
    │   │   ├── MessageProducer.java      # 接口：send(destination, Message)
    │   │   ├── MessageProducerImpl.java  # 装饰器：prepareHeaders + pre/post + withContext
    │   │   └── MessageProducerImplementation.java  # SPI：send + setMessageIdIfNecessary + withContext
    │   └── consumer/
    │       ├── MessageHandler.java       # extends Consumer<Message>
    │       ├── MessageConsumer.java      # 接口：subscribe + groupByChannel + close
    │       ├── MessageConsumerImpl.java  # 装饰器：ChannelMapping + DecoratedMessageHandlerFactory
    │       ├── MessageConsumerImplementation.java  # SPI
    │       ├── DecoratedMessageHandlerFactory.java   # 按 order 排序 → ChainBuilder 递归组装
    │       ├── MessageHandlerDecorator.java         # extends BiConsumer<SubscriberIdAndMessage, Chain> + getOrder()
    │       ├── MessageHandlerDecoratorChain.java    # 接口：invokeNext(SubscriberIdAndMessage)
    │       ├── MessageHandlerDecoratorChainBuilder.java  # 递归组装（private，已合并到 Factory）
    │       ├── DuplicateMessageDetector.java         # SPI：isDuplicate(consumerId, messageId) + doWithMessage
    │       ├── NoopDuplicateMessageDetector.java     # 默认实现（幂等默认关闭）：恒 false + 直调 callback
    │       ├── DuplicateMessageHandlerDecorator.java # 消息级去重装饰器（保留，不再默认装配）
    │       ├── PrePostHandleMessageHandlerDecorator.java
    │       ├── PrePostReceiveMessageHandlerDecorator.java
    │       ├── BuiltInMessageHandlerDecoratorOrder.java  # 内置 order 常量
    │       ├── ConsumeException.java
    │       └── MessageSubscription.java
    ├── commands/                       # ⚠️ 遗留代码：命令路由从未正式启用，不再发展，勿基于它做新功能
    ├── events/
    │   ├── EventMessageHeaders.java        # AGGREGATE_ID / AGGREGATE_TYPE / EVENT_TYPE
    │   ├── DomainEventNameMapping.java / DefaultDomainEventNameMapping.java
    │   ├── publisher/
    │   │   ├── DomainEventPublisher.java / DomainEventPublisherImpl.java
    │   │   └── ResultWithEvents.java       # 值对象：result + List<BaseDomainEvent>
    │   └── subscriber/
    │       ├── DomainEventHandler.java              # 三元组：aggregateType + eventClass + handler
    │       ├── DomainEventHandlers.java             # 聚合 List<DomainEventHandler>
    │       ├── DomainEventHandlersBuilder.java       # Builder：forAggregateType / onEvent / andForAggregateType
    │       ├── DomainEventEnvelope.java / DomainEventEnvelopeImpl.java
    │       ├── DomainEventDispatcher.java            # @PostConstruct initialize() + handleMessage（异常隔离 + handler 级幂等）
    │       ├── DomainEventDispatcherFactory.java / DispatcherFactoryBean.java
    │       └── BaseEventConsumer.java
    ├── adapter/                         # Bridge：easy-messaging ↔ SPI
    │   ├── EasyMessagingProducerAdapter.java
    │   ├── EasyMessagingConsumerAdapter.java
    │   └── ChannelProperties.java
    └── autoconfigure/
        ├── CqrsChannelsProperties.java   # 类名保持不变（历史命名）
        └── CqrsMessagingAutoConfiguration.java

七、消息装饰链

MessageHandlerDecoratorChainBuilder 递归组装装饰器责任链：

// 伪代码（核心逻辑）
private MessageHandlerDecoratorChain buildChain(List<MessageHandlerDecorator> handlers, Consumer<SubscriberIdAndMessage> terminal) {
    if (handlers.isEmpty()) {
        return terminal::accept;                    // 最终 handler
    }
    MessageHandlerDecorator head = handlers.get(0);
    List<MessageHandlerDecorator> tail = handlers.subList(1, handlers.size());
    return subscriberIdAndMessage -> head.accept(subscriberIdAndMessage, buildChain(tail, terminal));
}

默认装配的装饰器按 order 排序：

DuplicateMessageHandlerDecorator（消息级去重装饰器）类保留但不再默认装配：一个事件会被多个 handler 消费，消息级全局去重会误杀第 2..N 个 handler。幂等改为 handler 级，由 DomainEventDispatcher 直接与 DuplicateMessageDetector 协作（见 §7.1）。

7.1 消费语义：异常隔离 + handler 级幂等（v1.1.0）

一个事件会被多个业务场景（handler）消费，v1.1.0 之前 DomainEventDispatcher.handleMessage 用 stream().forEach 直调 handler，存在两个可靠性缺口：

设计上不改 DuplicateMessageDetector 接口，粒度下沉通过 dispatcher 内为每个 handler 构造新的 SubscriberIdAndMessage 实现：

for (DomainEventHandler handler : handlers.getHandlers()) {
    if (!handler.accept(message)) { continue; }
    String handlerKey = handler.getAggregateType() + ":" + handler.getEventClass().getSimpleName();
    try {
        duplicateMessageDetector.doWithMessage(
                new SubscriberIdAndMessage(eventDispatcherId + ":" + handlerKey, message),
                () -> handle(message, aggregateType, handler));
    } catch (Exception ex) {
        logger.warn("事件 handler 处理失败，跳过并继续后续 handler: handlerKey={}, messageId={}",
                handlerKey, message.getId(), ex);
    }
}

设计要点：

• 幂等 key：{dispatcherId}:{aggregateType}:{eventSimpleName} + messageId——同一事件被多个 handler 消费时互不误杀
• 异常隔离：per-handler try-catch，单 handler 失败仅记 warn 日志（含 handlerKey 与 messageId），继续执行后续 handler；消息不重投，失败补偿依赖日志告警
• 复用 SPI：Dispatcher 不感知幂等实现，Noop / 业务自定义均透明接入

幂等默认关闭：幂等是业务语义（唯一约束/状态机检查等可靠手段在业务层），框架不引入额外中间件（如 Redis）做默认实现，只保证消费链异常隔离 + SPI 挂载点。默认装配 NoopDuplicateMessageDetector（@ConditionalOnMissingBean，业务实现 SPI 注册 Bean 即覆盖，按 handler 级粒度生效）。

八、配置与后端

8.1 MQ 后端选择

注意：easy-event-bus 不直接创建后端——它通过 @AutoConfigureAfter(MessagingAutoConfiguration.class) 依赖 easy-messaging 的 BackendProducer/BackendConsumer Bean。后端切换由 easy-messaging 配置驱动。

8.2 SPI 可扩展点

九、使用示例

9.1 发布领域事件

// 1. 定义事件（extends BaseDomainEvent）
public class OrderPaidEvent extends BaseDomainEvent {
    private String orderId;
    private BigDecimal amount;
}

// 2. 聚合根发出 ResultWithEvents
public class OrderAggregate extends AggregateRoot {
    public ResultWithEvents<Order> pay() {
        // 业务逻辑
        OrderPaidEvent evt = new OrderPaidEvent(...);
        return new ResultWithEvents<>(this, evt);
    }
}

// 3. Application Service 发布
ResultWithEvents<Order> rwe = order.pay();
rwe.events.forEach(e ->
    domainEventPublisher.publish("com.example.Order", order.getId(), e, Collections.emptyMap())
);

9.2 订阅领域事件

// Spring Bean
@Component
public class OrderEventHandlerConfig {

    public OrderEventHandlerConfig(DomainEventDispatcherFactory factory) {
        DomainEventHandlers handlers = DomainEventHandlersBuilder
            .forAggregateType("com.example.Order")
            .onEvent(OrderPaidEvent.class, envelope -> {
                // envelope.getAggregateId() / envelope.getEvent() / envelope.getAggregateType()
                invoiceService.create(envelope.getEvent());
            })
            .onEvent(OrderShippedEvent.class, envelope -> {
                notificationService.send(envelope);
            })
            .andForAggregateType("com.example.Invoice")
            .onEvent(InvoiceCreatedEvent.class, envelope -> { ... })
            .build();

        factory.make("order-event-dispatcher", handlers);  // 启动时 @PostConstruct initialize()
    }
}

9.3 纯 SPI 后端（不依赖 easy-messaging）

// 自定义 ZeroMQ 后端，直接实现 SPI
public class ZeroMqProducerImpl implements MessageProducerImplementation {
    @Override public void send(Message message) { /* ZeroMQ publish */ }
}

// 业务层注入替换
@Configuration
public class ZeroMqConfig {
    @Bean public MessageProducerImplementation zeroMqProducer() {
        return new ZeroMqProducerImpl();
    }
}
// MessageProducerImpl 构造时自动注入新的 Implementation

十、关键设计决策

十一、测试覆盖

当前共 34 例（全绿），全部为纯 JVM 测试（零 MQ / 零 Spring 容器）：

可测程度 2/5（核心层 Spring-free 约 60%，Adapter/AutoConfiguration 属 Spring 集成测试范畴）：

• ✅ 已覆盖：Message/MessageBuilder/MessageProducerImpl/DefaultChannelMapping/NoopDuplicateMessageDetector/DomainEventDispatcher（异常隔离 + handler 级幂等）/ResultWithEvents（共 34 例）
• ❌ 未覆盖：DomainEventHandlersBuilder（纯 Builder，可补）、EasyMessagingProducerAdapter/EasyMessagingConsumerAdapter（需 easy-messaging BackendProducer mock）、CqrsMessagingAutoConfiguration（需 Spring Boot 测试容器）