事件总线与切面

soda 的自动化能力几乎全部来自两个 @Aspect：

切面	类	拦什么	干什么
CQRS 切面	`CqrsAroundHandler`	Command / Query / Event 的 `handle` 方法，`CommandBus.send`	上下文传递、递归保护、链路日志
仓储切面	`RepositoryEventAspect`	名字匹配 `*Repository` 的类型及其子类型的 `save` / `update` / `delete` / `operate`	取走领域事件，事务提交后发布

理解这两个切面，就理解了 soda 的大部分运行时行为。

CqrsAroundHandler

com.hibuka.soda.bus.interceptor.CqrsAroundHandler。它是 @Aspect 类，由 SpringBusAutoConfiguration 注册：

java

@Bean
@ConditionalOnMissingBean
public CqrsAroundHandler cqrsAroundHandler(EventBus eventBus, EventProperties eventProperties) {
    return new CqrsAroundHandler(eventBus, eventProperties);
}

它有两个 @Around 切点。

切点一：拦截所有 Handler

text

execution(* com.hibuka.soda.cqrs.command.CommandHandler+.handle(..)) ||
execution(* com.hibuka.soda.cqrs.query.QueryHandler+.handle(..)) ||
execution(* com.hibuka.soda.cqrs.event.EventHandler+.handle(..))

+ 是 AspectJ 的类型模式，表示「包括子类型」。所以你的 CreateProducerCommandHandler implements CommandHandler<...> 会被拦到。

这个切点做三件事。

1. 建立上下文。 根据第一个参数的类型，把上下文写进 DomainEventContext（ThreadLocal）：

java

if (firstArg instanceof BaseCommand) {
    BaseCommand cmd = (BaseCommand) firstArg;
    requestId = cmd.getRequestId();
    username = cmd.getUserName();

    DomainEventContext.setRequestId(requestId);
    DomainEventContext.setUserName(username);
    DomainEventContext.setJti(cmd.getJti());
    DomainEventContext.setAuthorities(cmd.getAuthorities());
    if (cmd.getCallerUid() != null) {
        DomainEventContext.setCallerUid(String.valueOf(cmd.getCallerUid()));
    }
    if (cmd.getHopCount() != null) {
        DomainEventContext.setHopCount(cmd.getHopCount());
    }
}

BaseQuery 分支只读 requestId 和 userName，不写上下文 —— 读链路不产生事件，不需要上下文。

AbstractDomainEvent 分支会恢复上下文，让 EventHandler 里发的新命令能继承链路信息。

2. 清理上下文。 在 finally 里：

java

} finally {
    // Clean up context to prevent memory leaks
    if (firstArg instanceof BaseCommand || firstArg instanceof AbstractDomainEvent) {
        DomainEventContext.clear();
    }
}

重要

这个 clear() 是必须的。DomainEventContext 是 ThreadLocal，线程池复用线程时不清理，上一个请求的 requestId 会串到下一个请求的事件里 —— 排障时会看到「A 用户的请求触发了 B 用户的数据变更」这种灵异现象。

3. 记录链路日志。 格式固定：

text

requestId:{},method:{}-{}, {}:{}, result:{}, duration:{}-{},username:{}

即：requestId:abc-123,method:com.example.demo.app.handler.CreateProducerCommandHandler-handle, CreateProducerCommand:{"name":"x"}, result:{"uid":"..."}, duration:12-0,username:alice。

两个耗时字段：第一个是方法执行耗时，第二个是日志序列化之后的耗时。

Redis 模式下的特殊分支

如果 soda.event.bus-type=redis，切面会跳过 EventHandler：

java

boolean streamEnabled = isStreamEnabled();
boolean isEventHandler = firstArg instanceof AbstractDomainEvent;
boolean isStreamConsumer = false;
boolean isAsyncStreamThread = Thread.currentThread().getName().startsWith("SimpleAsyncTaskExecutor");
if (isEventHandler && streamEnabled && !isStreamConsumer && !isAsyncStreamThread) {
    logger.info("[CqrsAroundHandler] Stream enabled, skip EventHandler on thread: {}", Thread.currentThread().getName());
    return null;
}

原因是：Redis 模式下事件已经写进 Stream，由 Redis 的消费线程负责反序列化和投递。本地再处理一次就会重复消费。

提示

isStreamConsumer 在当前代码里恒为 false —— 它是个预留开关。也就是说，判断依据实际上是「EventHandler 是不是在 SimpleAsyncTaskExecutor 线程上」。用默认的 cqrs-async- 线程池（ThreadPoolTaskExecutor）跑异步 EventHandler 时，这个线程名检查会失效，事件可能被跳过一次。

切点二：拦截 CommandBus.send

text

execution(* com.hibuka.soda.cqrs.command.CommandBus+.send(..))

这个切点做两件事。

1. 补全命令的上下文。 如果新命令的 requestId 是 null，从 DomainEventContext 里补：

java

if (cmd.getRequestId() == null) {
    String requestId = DomainEventContext.getRequestId();
    if (requestId != null) {
        cmd.setRequestId(requestId);
        cmd.setUserName(DomainEventContext.getUserName());
        cmd.setJti(DomainEventContext.getJti());
        cmd.setAuthorities(DomainEventContext.getAuthorities());
        String callerUid = DomainEventContext.getCallerUid();
        if (callerUid != null) {
            try {
                cmd.setCallerUid(callerUid);
            } catch (Exception ignored) {}
        }
    }
}

这就是「事件 → 命令」的链路能延续下去的原因：EventHandler 里 busFacade.sendCommand(newCmd) 时，新命令自动继承事件的上下文。

2. 异步递归保护。 每次 send 都会给命令的 hopCount 加一，超过 20 就抛异常：

java

Integer hopCount = DomainEventContext.getHopCount();
int currentHop = (hopCount != null) ? hopCount : 0;
if (currentHop > 20) {
    logger.error("[CqrsAroundHandler] Async recursion too deep! hopCount={}, command={}", currentHop, cmd.getClass().getSimpleName());
    throw new IllegalStateException("Async recursion too deep! hopCount=" + currentHop);
}
cmd.setHopCount(currentHop + 1);

重要

20 跳的上限是异步的保护。同步递归由 BusFacade 的 MAX_RECURSION_DEPTH = 10 保护。两套机制独立生效。详见递归保护。

3. 发布命令产生的事件。 这是切面二的第三件事，容易被忽略：

java

if (result instanceof DomainEvents) {
    for (AbstractDomainEvent event : ((DomainEvents) result).getDomainEvents()) {
        sycBaseInfo((BaseCommand) args[0], event);
        if (streamEnabled) {
            // Redis 模式：跳过本地发布，交给仓储切面
        } else {
            eventBus.publish(event);
        }
    }
} else if (result instanceof DomainEvent) {
    // 命令直接返回一个领域事件
    ...
}

也就是说：如果 CommandHandler 返回的聚合根还带着未发布的领域事件，切面会直接发布它们。

注意

这会和仓储切面重复发布吗？不会 —— 仓储切面在 save 时就把事件取走并 clearDomainEvents() 了。等切面二检查返回值时，事件列表已经空了。

但如果你在 CommandHandler 里 save 之后又手动 addPendingEvent（或者不调 save 只调 update），就可能出现事件被发两次的情况。规则是：事件只由仓储切面发布，Handler 里不要手动 eventBus.publish。

RepositoryEventAspect

com.hibuka.soda.event.spring.RepositoryEventAspect。它由 RepositoryEventAutoConfiguration 注册：

java

@Configuration
@EnableAspectJAutoProxy
@ConditionalOnBean(EventBus.class)
@AutoConfigureAfter(name = {
        "com.hibuka.soda.event.spring.SpringEventBusAutoConfiguration",
        "com.hibuka.soda.event.redis.RedisEventBusAutoConfiguration"
})
public class RepositoryEventAutoConfiguration {
    @Bean
    public RepositoryEventAspect repositoryEventAspect(EventBus eventBus, ObjectProvider<CommandContext> commandContextProvider) {
        return new RepositoryEventAspect(eventBus, commandContextProvider);
    }
}

@ConditionalOnBean(EventBus.class) 意味着：没有 EventBus Bean 时，这个切面根本不注册，仓储调用就只是普通的方法调用。

切点

text

execution(* *..*Repository+.save(..)) ||
execution(* *..*Repository+.update(..)) ||
execution(* *..*Repository+.delete(..)) ||
execution(* *..*Repository+.operate(..))

注意

这个表达式有三个关键点，每一个都是坑：

*..*Repository+ 匹配的是名字以 Repository 结尾的类型，以及它的子类型。所以仓储接口叫 ProducerWriteRepository 就够了 —— 实现类叫 ProducerRepoImpl 也会被拦到。但接口和实现类的名字里都没有 Repository 时（比如 ProducerStore），切点不命中，事件静默丢失。
方法名必须是 save / update / delete / operate。改名叫 persist 或 insert，同样静默失效。
*..* 表示任意包。这意味着你的任何模块里凡是叫 XxxRepository 的类都会被匹配上。切面会检查第一个参数是不是 AbstractAggregateRoot 的子类，不是就跳过 —— 但如果你有个恰好叫 OrderRepository 的工具类，里面有个 save(String) 方法，它也会被这个切面看到（只是被跳过，不会有影响）。

排查口诀：事件没发出去，先看仓储类型名里有没有 Repository，再看方法名对不对。

执行流程

java

public Object aroundSave(ProceedingJoinPoint joinPoint) throws Throwable {
    Object result = joinPoint.proceed();          // ① 先执行真正的 save
    Object[] args = joinPoint.getArgs();

    if (args != null && args.length > 0 && args[0] instanceof AbstractAggregateRoot) {
        AbstractAggregateRoot agg = (AbstractAggregateRoot) args[0];
        List<AbstractDomainEvent> events = List.copyOf(agg.getDomainEvents());

        if (!events.isEmpty()) {
            agg.clearDomainEvents();                // ② 取走事件并清空
            enrichEventsWithContext(events);        // ③ 补上下文

            if (TransactionSynchronizationManager.isSynchronizationActive()) {
                // ④ 有事务：注册 afterCommit 回调
                TransactionSynchronizationManager.registerSynchronization(
                        new TransactionSynchronization() {
                            @Override
                            public void afterCommit() {
                                publishEvents(events);
                            }
                        });
            } else {
                // ⑤ 没事务：立即发布
                publishEvents(events);
            }
        }
    }
    return result;
}

关键点：

先执行 save，再取事件。 顺序不能反 —— save 失败抛异常，事件就不会被取走，也就不会发布。
List.copyOf 复制一份快照，然后清空聚合里的列表。这样即使事件在发布前被别处修改，也不影响。
有事务走 afterCommit，无事务立即发布。

事务回调

java

if (TransactionSynchronizationManager.isSynchronizationActive()) {
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
        @Override
        public void afterCommit() {
            publishEvents(events);
        }
    });
} else {
    publishEvents(events);
}

isSynchronizationActive() 判断的是「当前线程是否在事务同步中」，也就是事务是否已经开始。Spring 管理的 @Transactional 方法里它是 true。

重要

事务回滚时 afterCommit 不会执行 —— 事件不会被发布。这就是「事务提交后才发事件」的实现：不会出现「业务失败了但事件发出去」的幻象。

反过来，如果你不加 @Transactional，事件就在 save 返回后立即发布。这时候 save 成功了但后续逻辑失败了，事件已经发出去了 —— 数据和事件就不一致了。所以写链路请务必开事务。

上下文补全

enrichEventsWithContext 有两级优先级：

java

// Priority 1: CommandContext (Legacy)
if (ctx != null) {
    events.forEach(e -> {
        e.setRequestId(ctx.getRequestId());
        e.setJti(ctx.getJti());
        String[] authoritiesArray = ctx.getAuthorities();
        if (authoritiesArray != null) {
            e.setAuthorities(String.join(",", authoritiesArray));
        }
        e.setUserName(ctx.getUserName());
        ...
    });
}

// Priority 2: DomainEventContext (New, supports hopCount and context propagation)
String requestId = DomainEventContext.getRequestId();
if (requestId != null) {
    events.forEach(e -> {
        if (e.getRequestId() == null) {
            e.setRequestId(requestId);
            ...
        }
        // Always update hopCount from ThreadLocal if available
        Integer hopCount = DomainEventContext.getHopCount();
        if (hopCount != null) {
            e.setHopCount(hopCount);
        }
    });
}

CommandContext 是旧机制，authorities 是 String[]，用逗号 join。
DomainEventContext 是新机制，多了 hopCount。它的字段只在为空时才填（if (e.getRequestId() == null)），但 hopCount 是无条件覆盖 —— 因为它要支持异步递归保护。

注意 CommandContext 优先级更高：它一旦存在，就会无条件覆盖事件的上下文字段。

发布

java

private void publishEvents(List<AbstractDomainEvent> events) {
    for (AbstractDomainEvent event : events) {
        try {
            eventBus.publish(event);
        } catch (Exception e) {
            log.error("Failed to publish domain event: {}", event, e);
        }
    }
}

提示

单个事件发布失败不会影响其他事件，也不会让事务回滚。 异常被 catch 住只记日志。这意味着：事件投递失败时，你的业务数据已经提交了，但下游不知道。跨服务场景下这需要靠 Redis 模式的死信流或补偿任务兜底。

两种事件总线

EventBus 接口有三个方法：

java

public interface EventBus {
    void publish(DomainEvent event) throws BaseException;
    void subscribe(Class<? extends DomainEvent> eventType, EventHandler handler) throws BaseException;
    void unsubscribe(Class<? extends DomainEvent> eventType, EventHandler handler) throws BaseException;
}

SimpleEventBus

进程内实现。soda.event.bus-type 未配置或为 simple 时启用：

java

@Configuration
@ConditionalOnProperty(name = "soda.event.bus-type", havingValue = "simple", matchIfMissing = true)
public class SimpleEventBusAutoConfiguration {
    @Bean
    @ConditionalOnMissingBean
    public EventBus simpleEventBus(List<EventHandler<? extends DomainEvent>> eventHandlers) {
        return new SimpleEventBus(eventHandlers);
    }
}

内部结构是一个 Map<Class<? extends DomainEvent>, List<EventHandler>>，List 用 CopyOnWriteArrayList，所以一个事件类型可以有多个 Handler。

发布时的类型查找是沿着类型链向上走的：

java

public void publish(DomainEvent event) throws BaseException {
    Class<?> eventType = event.getClass();
    while (eventType != null) {
        List<EventHandler> eventHandlers = handlers.get(eventType);
        if (eventHandlers != null && !eventHandlers.isEmpty()) {
            for (EventHandler handler : eventHandlers) {
                handler.handle(event);
            }
        }

        // Check all interfaces implemented by this class
        for (Class<?> interfaceType : eventType.getInterfaces()) {
            if (DomainEvent.class.isAssignableFrom(interfaceType)) {
                List<EventHandler> interfaceHandlers = handlers.get(interfaceType);
                ...
            }
        }

        eventType = eventType.getSuperclass();
    }
}

也就是说，Handler implements EventHandler<AbstractDomainEvent> 会收到所有领域事件；Handler implements EventHandler<ProducerEvent> 只收到 ProducerEvent 及其子类。

技巧

写一个 EventHandler<DomainEvent> 当作「全局审计 Handler」是很方便的用法 —— 它能捕获所有事件。

RedisStreamEventBus

跨服务实现。soda.event.bus-type=redis 时启用，需要 soda-event-redis 模块和 classpath 上有 RedisTemplate。

java

@Configuration
@ConditionalOnClass(RedisTemplate.class)
@ConditionalOnProperty(name = "soda.event.bus-type", havingValue = "redis")
public class RedisEventBusAutoConfiguration {
    @Bean
    @Primary
    @ConditionalOnBean(RedisConnectionFactory.class)
    public EventBus redisStreamEventBus(...) { ... }
}

它是 @Primary 的 —— 有 Redis 时，它会覆盖掉 SimpleEventBus。

它基于 Redis Stream，提供了：

能力	说明
消费者组	`XGROUP CREATE`，同一组内多个消费者分摊
消费确认	`XACK`，处理成功才确认
批量拉取	`XREADGROUP`，`batchSize` 控制单次条数
并发消费	`concurrency` 个消费线程，`consumerName` 会加 `-0`、`-1` 后缀
失败重试	最多 `maxRetries` 次，延迟 `initialRetryDelay * 2^n`
死信流	重试耗尽后写入 `deadLetterStream`
幂等	基于 `IdempotencyService`，按 `eventId` 去重
定期清理	`@Scheduled(fixedRate = 3600000)` 清理过期的幂等状态

发布时的 payload：

java

Map<String, String> entry = new HashMap<>(3);
String eventJson = serializer.toJson(event);
entry.put("event", eventJson);
entry.put("type", event.getClass().getName());
if (context != null) {
    entry.put("context", serializer.toJson(context));
}
streamOps.add(streamKey, entry);

type 存的是事件类的全限定名，消费端靠它反序列化。

Redis 总线的 publish 也做了事务处理：

java

if (TransactionSynchronizationManager.isActualTransactionActive()) {
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
        @Override
        public void afterCommit() {
            publishAction.run();
        }
    });
} else {
    publishAction.run();
}

提示

注意这里用的是 isActualTransactionActive()，而仓储切面用的是 isSynchronizationActive()。语义相近但不完全相同 —— 前者判断「有没有真实的事务」，后者判断「事务同步机制是否激活」。实际使用中差别不大。

spring 这一档其实不存在

EventProperties.busType 的默认值是 "spring"，文档里也常写 spring / simple / redis 三选一。但在 17.0.1 里：

soda-event-spring 模块已经被删除
没有任何 EventBus 实现类的条件是 havingValue = "spring"

所以配 soda.event.bus-type=spring 会导致没有任何 EventBus Bean。后果是：

RepositoryEventAutoConfiguration 因为 @ConditionalOnBean(EventBus.class) 不注册 → 仓储切面失效 → 领域事件永远发不出去
SpringBusAutoConfiguration.cqrsAroundHandler 需要 EventBus 参数 → Bean 创建失败

注意

生产环境请用 simple 或 redis，不要用 spring。

更稳妥的做法是显式配置 soda.event.bus-type=simple。不配置时 SimpleEventBusAutoConfiguration 的 matchIfMissing = true 会生效（走 Simple），但显式写出来更清楚 —— 顺便也避开了 EventProperties 里那个 "spring" 默认值带来的误导。

自动装配清单

soda-core-base 的 META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports：

text

com.hibuka.soda.bus.configuration.SpringBusAutoConfiguration
com.hibuka.soda.bus.configuration.SimpleEventBusAutoConfiguration
com.hibuka.soda.event.spring.RepositoryEventAutoConfiguration

soda-event-redis 的同一个文件：

text

com.hibuka.soda.event.redis.RedisEventBusAutoConfiguration

SpringBusAutoConfiguration 注册的 Bean：

Bean	条件	说明
`commandBus`	`@ConditionalOnMissingBean`	`SimpleCommandBus`
`queryBus`	`@ConditionalOnMissingBean`	`SimpleQueryBus`
`cqrsAsyncExecutor`	`@ConditionalOnMissingBean(name = ...)`	`ThreadPoolTaskExecutor`
`busFacade`	`@ConditionalOnMissingBean`	`BusFacade`
`cqrsAroundHandler`	`@ConditionalOnMissingBean`	CQRS 切面
`eventProperties`	`@ConditionalOnMissingBean`	`soda.event.*` 配置

技巧

所有关键 Bean 都是 @ConditionalOnMissingBean。这就是扩展点：你自己定义一个同名的 Bean，框架的就会被跳过。 详见扩展点。

两个切面的执行顺序

一次完整的写请求里，两个切面是这样嵌套的：

text

BusFacade.sendTransactCommand(command)
  │
  ├─ CommandBus.send(command)
  │    │
  │    ├─ CqrsAroundHandler.handleEventBus      ← 外层
  │    │    ├─ 补命令上下文
  │    │    ├─ hopCount 检查，+1
  │    │    ├─ proceed()
  │    │    │    └─ SimpleCommandBus.send → 找到 CommandHandler → handle(command)
  │    │    │         │
  │    │    │         └─ CqrsAroundHandler.handleExecutor   ← 内层
  │    │    │              ├─ 写 DomainEventContext
  │    │    │              ├─ proceed()
  │    │    │              │    ├─ domain.create() → addPendingEvent
  │    │    │              │    └─ producerWriteRepository.save(domain)
  │    │    │              │         └─ RepositoryEventAspect.aroundSave
  │    │    │              │              ├─ proceed() → 真正的 insert
  │    │    │              │              ├─ 取事件、清空、补上下文
  │    │    │              │              └─ 注册 afterCommit
  │    │    │              ├─ clear() DomainEventContext
  │    │    │              └─ 记链路日志
  │    │    │
  │    │    └─ 检查返回值：DomainEvents / DomainEvent → publish
  │    │
  │    └─ 返回
  │
  └─ 事务提交
       └─ afterCommit → eventBus.publish(event)
            └─ ProducerEventHandler.handle(event)
                 └─ CqrsAroundHandler.handleExecutor（又一次）

注意仓储切面在 CQRS 切面里面。所以 save 的时候，DomainEventContext 还在（还没被 clear），事件能拿到上下文。

排查清单

事件没发出去，按这个顺序查：

仓储类型的名字里有 Repository 吗？（接口或实现类任一匹配都行）
方法名是 save / update / delete / operate 吗？
第一个参数是 AbstractAggregateRoot 的子类吗？
聚合根的业务方法调了 addPendingEvent 吗？（protected，外部调不了）
soda.event.bus-type 是不是被配成了 spring？
EventBus Bean 存在吗？启动日志里搜 [SimpleEventBusAutoConfiguration] Creating SimpleEventBus instance
事务是不是回滚了？afterCommit 只在提交成功时执行。

启动日志里这几行能直接确认注册情况：

text

[SimpleCommandBus] Constructor called, handlers size: 1
[SimpleCommandBus] Registered handler for command: com.example.demo.CreateProducerCommand
[SimpleEventBus] Constructor called, handlers size: 1
[SimpleEventBus] Registered 1 event handler types
[CqrsAroundHandler] Bean created!

下一步

自己实现事件总线、序列化器、仓储：扩展点
上下文是怎么跨线程、跨服务传递的：上下文传递
递归是怎么被挡住的：递归保护

事件总线与切面#

CqrsAroundHandler#

切点一：拦截所有 Handler#

Redis 模式下的特殊分支#

切点二：拦截 CommandBus.send#

RepositoryEventAspect#

切点#

执行流程#

事务回调#

上下文补全#

发布#

两种事件总线#

SimpleEventBus#

RedisStreamEventBus#

spring 这一档其实不存在#

自动装配清单#

两个切面的执行顺序#

排查清单#

下一步#

事件总线与切面

CqrsAroundHandler

切点一：拦截所有 Handler

Redis 模式下的特殊分支

切点二：拦截 CommandBus.send

RepositoryEventAspect

切点

执行流程

事务回调

上下文补全

发布

两种事件总线

SimpleEventBus

RedisStreamEventBus

spring 这一档其实不存在

自动装配清单

两个切面的执行顺序

排查清单

下一步