easy-event-bus 使用文档

easy-event-bus 提供领域事件发布/订阅（DomainEventPublisher + DomainEventDispatcher）完整链路。业务层只需定义 BaseDomainEvent、注册 DomainEventHandler，后端 MQ（RocketMQ / Redis / JDBC / ZeroMQ / Kafka）通过 easy-messaging SPI 自动切换，无需硬编码。组件支持消息拦截器链、幂等检测、重复消息过滤、统一头（AGGREGATE_ID / AGGREGATE_TYPE / EVENT_TYPE）。

更名记录：组件由 easy-cqrs 更名为 easy-event-bus（框架走 CQS 范式，专注事件总线）。Java 包名 io.matrix.cqrs、类名与配置前缀 cqrs.* 保持不变，消费方 import 与已上线配置零影响。原命令路由（CommandProducer / commands 包）从未正式启用且不再发展，本文档不再覆盖。

• 1. 引入依赖
• 2. 选择 MQ 后端
• 3. 配置
• 4. 定义领域事件 Event
• 5. 发布领域事件 DomainEventPublisher
• 6. 订阅领域事件 DomainEventDispatcher
• 7. 拦截器链
• 8. 幂等与重复消息检测
• 9. 配置项详解
• 10. 注意事项
• 11. 常见问题

1. 引入依赖

<!-- 事件总线核心（必需，compile scope） -->
<dependency>
    <groupId>io.matrix</groupId>
    <artifactId>easy-event-bus</artifactId>
    <version>${revision}</version>
</dependency>

<!-- 依赖 easy-messaging（compile scope，easy-event-bus 自动传递引入） -->

<!-- MQ 后端（optional scope，按需引入，详见 §2） -->
<dependency>
    <groupId>io.matrix</groupId>
    <artifactId>easy-messaging-rocketmq</artifactId>  <!-- 或 easy-messaging-redis / jdbc / zeromq / kafka -->
</dependency>

easy-event-bus 传递依赖 easy-messaging（compile scope），但 easy-messaging 各 MQ backend 均为 optional——没有后端就没有 ProducerImplementation Bean，启动时需保证至少引入一个 backend。

2. 选择 MQ 后端

推荐策略：生产优先 RocketMQ，POC / 中小项目 Redis，测试用 JDBC（内嵌 H2 + 锁表）。

自动配置顺序（AGENTS.md 强制约定）：

easy-messaging backend 使用 @AutoConfigureBefore(MessagingAutoConfiguration.class)；easy-event-bus 使用 @AutoConfigureAfter(MessagingAutoConfiguration.class)。保证 backend Producer/Consumer Bean 先于事件总线门面创建。

3. 配置

3.1 YAML 最小配置

# application.yml（配置前缀 cqrs.*，属历史命名保留）
cqrs:
  # 通道配置：channel 名 → 实际 MQ topic / queue / table 的映射
  # 没配时使用 DefaultChannelMapping：channel 名 = 后端名
  channels:
    # 事件通道（领域事件总线）
    event:
      backend: rocketmq
      topic: matrix-event
      group: matrix-ev-consumer
    # 可以按业务域分多通道
    order-event:
      backend: redis
      topic: order-events          # Redis Pub/Sub channel 名
      group: ""                    # Redis 不需要 group

# 后端 MQ 连接配置（标准 Spring Boot）
rocketmq:
  name-server: 127.0.0.1:9876
spring:
  redis:
    host: localhost
    port: 6379

3.2 后端自动选择

当 easy-messaging 容器中只有一个可用 Producer/Consumer 时，事件总线默认用它。配置里 backend: 字段指定显式选择。

4. 定义领域事件 Event

import io.matrix.kernel.event.BaseDomainEvent;   // kernel-client 提供

// Event：继承 BaseDomainEvent
// BaseDomainEvent 自带 headers：AGGREGATE_ID / AGGREGATE_TYPE / EVENT_TYPE
public class OrderCreatedEvt extends BaseDomainEvent {
    private String orderId;
    private String customerId;
    private BigDecimal amount;
    // constructor / getter / builder ...
}

语义：Event 是事实（“X 已经发生”），可以被多个 DomainEventHandler 并行订阅。框架走 CQS 范式：命令（Cmd）经本地 CommandHandler 同步执行，跨服务集成一律走领域事件。

5. 发布领域事件 DomainEventPublisher

@Service
public class OrderService {

    @Autowired private DomainEventPublisher eventPublisher;

    public void completePay(String orderId) {
        // 执行业务...
        order.markPaid();

        // 发布领域事件
        OrderPaidEvt evt = new OrderPaidEvt(orderId, customerId, amount);
        eventPublisher.publish(evt, "event");
        // 或 eventPublisher.publish(evt) 默认通道名 "event"
    }
}

// 聚合根一次性返回多事件（推荐）
@Service
public class OrderAggregateService {

    @Autowired private DomainEventPublisher eventPublisher;

    public Result completePay(String orderId) {
        ResultWithEvents<Order> result = aggregate.completePay(orderId);

        // 批量发布
        result.events().forEach(evt -> eventPublisher.publish(evt, "event"));
        return Result.ok(result.value());
    }
}

6. 订阅领域事件 DomainEventDispatcher

// 方式一（推荐）：继承 BaseEventConsumer + DomainEventHandlersBuilder 声明 handler
@Component
public class OrderEventConsumer extends BaseEventConsumer {

    @Autowired private InventoryService inventoryService;

    @Override
    public DomainEventHandlers domainEventHandlers() {
        return DomainEventHandlersBuilder
            .forAggregateClass(Order.class)
            .onEvent(OrderCreatedEvt.class, envelope -> {
                // envelope.getEvent() 已反序列化为强类型事件
                inventoryService.deduct(envelope.getEvent().getOrderId());
            })
            .onEvent(OrderPaidEvt.class, envelope -> {
                notificationService.notifyPaid(envelope.getAggregateId());
            })
            .andForAggregateClass(Stock.class)      // 可继续声明其他聚合的事件
            .onEvent(StockAdjustedEvt.class, this::sync)
            .build();
    }
}

// 方式二：DomainEventHandlersBuilder 手动构建 + FactoryBean 注册（单元测试/动态路由）
DomainEventHandlers handlers = DomainEventHandlersBuilder
    .forAggregateType("io.matrix.Order")
    .onEvent(OrderCreatedEvt.class, evt -> { ... })
    .build();

消费语义（一个事件可被多个 handler 消费）：

以上语义在 DomainEventDispatcher 内实现，对业务 handler 透明，无需自行 try-catch 或做消息级去重。

DomainEventDispatcher 在 @PostConstruct 里执行 initialize()：扫描容器中所有 BaseEventConsumer Bean → 按通道名注册 → 组装 MessageHandlerDecoratorChain。单元测试需手动调 dispatcher.initialize()。

7. 拦截器链

装饰器链按 getOrder() 升序递归组装，chain.invokeNext() 之后是 handler 实际调用：

// 自定义装饰器：实现 accept(SubscriberIdAndMessage, Chain) + getOrder()
@Component
public class LoggingDecorator implements MessageHandlerDecorator {

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

    @Override
    public void accept(SubscriberIdAndMessage subscriberIdAndMessage, MessageHandlerDecoratorChain chain) {
        Message message = subscriberIdAndMessage.getMessage();
        log.info("Received message: {}", message.getId());
        try {
            chain.invokeNext(subscriberIdAndMessage);   // 必须调用，链才会继续执行
        } finally {
            log.info("Processed message: {}", message.getId());
        }
    }

    @Override
    public int getOrder() {
        return Ordered.HIGHEST_PRECEDENCE + 10;
    }
}

// 日志/跟踪等横切逻辑也可以用更细粒度的 MessageInterceptor（Spring Bean 自动装配为装饰器）
public interface MessageInterceptor {
    default void preSend(Message message) { }
    default void postSend(Message message, Exception e) { }
    default void preReceive(Message message) { }
    default void preHandle(String subscriberId, Message message) { }
    default void postHandle(String subscriberId, Message message, Throwable throwable) { }
    default void postReceive(Message message) { }
}

框架内置装饰器（DecoratedMessageHandlerFactory 按 order 排序递归组装）：

DuplicateMessageHandlerDecorator（消息级去重装饰器）类保留但不再默认装配：一个事件会被多个 handler 消费，消息级去重会误杀第 2..N 个 handler，幂等已下沉为 handler 级粒度（见 §8）。

8. 幂等与重复消息检测

框架默认不开启幂等（NoopDuplicateMessageDetector 兜底）。幂等是业务语义——同一事件，状态覆盖类 handler 天然幂等，扣款类 handler 必须防重——框架无法替业务统一决定，也不引入额外中间件（如 Redis）做默认实现。框架只保证两件事：消费链异常隔离（§6）+ 幂等 SPI 挂载点。

业务确有防重需求时，优先在业务层用可靠手段实现（数据库唯一约束、状态机检查、版本号等）；确需消息级去重时，实现 DuplicateMessageDetector 注册为 Bean 即可，框架自动按 handler 级粒度生效（key = dispatcherId:聚合类型:事件名 + messageId，多 handler 场景互不误杀）。

DuplicateMessageDetector SPI（真实签名）：

public interface DuplicateMessageDetector {
    boolean isDuplicate(String consumerId, String messageId);
    /** 不重复时执行 callback（幂等判定与执行在实现内闭环） */
    void doWithMessage(SubscriberIdAndMessage subscriberIdAndMessage, Runnable callback);
}

自定义实现示例（Redis SET NX 占位即标记）：

@Component
public class RedisDuplicateMessageDetector implements DuplicateMessageDetector {

    @Autowired private StringRedisTemplate redis;

    @Override
    public boolean isDuplicate(String consumerId, String messageId) {
        return !Boolean.TRUE.equals(redis.opsForValue()
                .setIfAbsent("eventbus:dedup:" + consumerId + ":" + messageId, "1", Duration.ofHours(24)));
    }

    @Override
    public void doWithMessage(SubscriberIdAndMessage subscriberIdAndMessage, Runnable callback) {
        if (isDuplicate(subscriberIdAndMessage.getSubscriberId(), subscriberIdAndMessage.getMessage().getId())) {
            return;                                 // 已处理过，跳过
        }
        callback.run();
    }
}

语义提醒：SET NX 占位即标记（标记在 handler 执行前完成、失败不回滚）是 at-most-once——handler 失败该消息不会再来，与 MQ 重投机制相悖。若需 at-least-once（失败可重试），改为”callback 执行成功后才标记”。

9. 配置项详解

easy-event-bus 的配置集中在 cqrs.channels.*（前缀保持历史命名），每个通道可独立指定 backend / topic / group。

cqrs:
  channels:
    <channel-name>:
      backend: rocketmq | redis | jdbc | zeromq | kafka   # 必填，选择后端
      topic: string                                       # 对应后端的 topic / channel / table 名
      group: string                                       # Consumer 消费组（RocketMQ/Kafka 必需，Redis 可选，JDBC 可选）
  # event.dispatcher: any                                 # 设置后启用 DispatcherFactoryBean（扫描所有 BaseEventConsumer 自动组装分发器）

10. 注意事项

11. 常见问题

Q: 可以不用 MQ 后端吗？ A: 不行。easy-event-bus 100% 依赖 easy-messaging SPI，没有 ProducerImplementation / ConsumerImplementation Bean 会直接启动失败。如果要进程内消息，建议用 Guava EventBus 或 Spring ApplicationEvent。

Q: 消息序列化怎么选？ A: MessageBuilder 默认用 JsonMapper（kernel-common）。BaseDomainEvent 需有无参构造器 + getter/setter。

Q: 同一个 Event 被多个 Handler 订阅怎么路由？ A: DomainEventDispatcher 内部维护 Map<EventClass, List<DomainEventHandler>>，handleMessage 遍历所有 handler 执行。没有顺序保证（除非显式加 @Order）。

Q: EventHandler 抛异常会怎么样？ A: 框架做异常隔离：只记 warn 日志（含 handlerKey 与 messageId），继续执行同消息的后续 handler，消息不重投。防重需求见 §8。

Q: 同一事件被多个 handler 消费，幂等会互相影响吗？ A: 不会。幂等 key 按 handler 粒度隔离（dispatcherId:聚合类型:事件名 + messageId），业务注册 DuplicateMessageDetector 后生效，handler A 已处理不影响 handler B 的正常执行。

Q: Channel 名和 Topic 名不一样会怎么样？ A: 正确——通道名是事件总线层的抽象（"event" / "order-event"），topic 是 MQ 后端的实际命名。cqrs.channels.<name>.topic 做了映射。没配置 DefaultChannelMapping 会用 channel 名 = topic 名。

Q: 为什么配置前缀是 cqrs 而组件叫 easy-event-bus？ A: 前缀属于历史命名，更名时保持不变以兼容已上线配置与代码；包名 io.matrix.cqrs 同理。详见 design.md 更名记录。