请求是怎么跑完的

框架的价值不在于给了你几个基类，而在于把散落在各处的横切逻辑收进了一条确定的链路。这一页把这条链路的每一段拆开讲。

我们跟着一句 busFacade.sendCommand(command) 往前走。

全景

text

HTTP 请求
   │
   ▼
[1] Controller                    你的代码
   │  组装命令，把请求头里的身份信息塞进命令
   ▼
[2] BusFacade.sendCommand(cmd)    框架入口
   │
   ▼
[3] CqrsAroundHandler @Around     ── CommandBus.send(..)
   │  · 上下文回填（命令没带 requestId 时，从 ThreadLocal 补）
   │  · hopCount 检查与 +1
   │  · 调用真正的总线
   ▼
[4] CommandBusImpl.send(cmd)      总线按类型找 CommandHandler
   │
   ▼
[5] CqrsAroundHandler @Around     ── CommandHandler.handle(..)
   │  · 把命令里的上下文写进 DomainEventContext（ThreadLocal）
   │  · 执行你的 CommandHandler
   │  · finally 里清理 ThreadLocal
   ▼
[6] CommandHandler.handle(cmd)    你的代码
   │  · 校验 → 聚合根改状态 → addPendingEvent
   ▼
[7] RepositoryEventAspect @Around ── *Repository+.save(..)
   │  · 保存聚合根
   │  · 取出挂起的领域事件，注入上下文
   │  · 有事务 → afterCommit 发布；无事务 → 立即发布
   ▼
[8] EventBus.publish(event)       ── Simple / Redis
   │
   ▼
[9] CqrsAroundHandler @Around     ── EventHandler.handle(..)
   │  · 从事件里恢复上下文（事件带着 requestId 过来的）
   ▼
[10] EventHandler.handle(event)   你的代码

下面逐段看细节。

[1] Controller —— 上下文的第一站

框架不强制你用任何 Web 层的基类。BaseCommand 上的上下文字段需要有人填，最常见的做法是加一个切面：

java

@Aspect
@Component
public class ControllerAroundHandler {

    @Around("execution(* ..*Controller..*(..))")
    public Object around(ProceedingJoinPoint jp) throws Throwable {
        HttpServletRequest request =
                ((ServletRequestAttributes) RequestContextHolder.currentRequestAttributes())
                        .getRequest();

        String requestId = request.getHeader(BaseConstant.SODA_REQUEST_ID);
        if (StringUtils.isBlank(requestId)) {
            requestId = request.getHeader(BaseConstant.X_REQUEST_ID);
        }
        if (StringUtils.isBlank(requestId)) {
            requestId = UUID.randomUUID().toString();
        }
        String userName = request.getHeader(BaseConstant.SODA_USER_NAME);

        // 写进 MDC，日志自动带上
        MDC.put("requestId", requestId);
        MDC.put("username", userName);

        Object[] args = jp.getArgs();
        if (args.length > 0 && args[0] instanceof BaseRequest) {
            ((BaseRequest) args[0]).setRequestInfo(requestId, userName);
        }
        return jp.proceed();
    }
}

BaseConstant 里定义了框架认的请求头：

常量	请求头
`SODA_REQUEST_ID`	`Soda-Request-Id`
`SODA_USER_NAME`	`Soda-User-Name`
`X_REQUEST_ID`	`X-Request-Id`
`X_UID`	`X-UID`
`X_USER_NAME`	`X-User-Name`
`X_AUTHORITIES`	`X-Authorities`
`X_JTI`	`X-Jti`
`X_CALLER_UID`	`X-Caller-Uid`
`X_TENANT_ID`	`X-Tenant-Id`

提示

BaseConstant 里还有 SCADA = "scada"，是框架内部预留的一个标识常量。

[3] 命令总线切面 —— 上下文回填与 hopCount

CqrsAroundHandler 拦截 CommandBus.send(..)：

java

@Around("execution(* com.hibuka.soda.cqrs.command.CommandBus+.send(..))")
public Object handleEventBus(ProceedingJoinPoint joinPoint) throws Throwable {
    Object[] args = joinPoint.getArgs();
    if (args.length > 0 && args[0] instanceof BaseCommand) {
        BaseCommand cmd = (BaseCommand) args[0];

        // 上下文回填：命令自己没带 requestId，就从 ThreadLocal 拿
        if (cmd.getRequestId() == null) {
            String requestId = DomainEventContext.getRequestId();
            if (requestId != null) {
                cmd.setRequestId(requestId);
                cmd.setUserName(DomainEventContext.getUserName());
                // ... jti / authorities / callerUid
            }
        }

        // 递归保护
        Integer hopCount = DomainEventContext.getHopCount();
        int currentHop = (hopCount != null) ? hopCount : 0;
        if (currentHop > 20) {
            throw new IllegalStateException("Async recursion too deep! hopCount=" + currentHop);
        }
        cmd.setHopCount(currentHop + 1);
    }
    return joinPoint.proceed();
}

上下文回填解决的是这个场景：你在事件消费者里发了一个新命令，这个新命令没有 requestId，但当前线程的 ThreadLocal 里还留着上一个事件带过来的 requestId。回填让它自动接上链路。

hopCount 是异步递归保护，见递归保护。

[5] Handler 切面 —— 上下文注入与清理

CqrsAroundHandler 还拦截所有 Handler：

java

@Around("execution(* com.hibuka.soda.cqrs.command.CommandHandler+.handle(..)) || " +
        "execution(* com.hibuka.soda.cqrs.query.QueryHandler+.handle(..)) || " +
        "execution(* com.hibuka.soda.cqrs.event.EventHandler+.handle(..))")
public Object handleExecutor(ProceedingJoinPoint point) throws Throwable {
    Object firstArg = point.getArgs()[0];

    if (firstArg instanceof BaseCommand) {
        BaseCommand cmd = (BaseCommand) firstArg;
        DomainEventContext.setRequestId(cmd.getRequestId());
        DomainEventContext.setUserName(cmd.getUserName());
        DomainEventContext.setJti(cmd.getJti());
        DomainEventContext.setAuthorities(cmd.getAuthorities());
        // ...
    } else if (firstArg instanceof AbstractDomainEvent) {
        AbstractDomainEvent event = (AbstractDomainEvent) firstArg;
        DomainEventContext.setRequestId(event.getRequestId());
        // 从事件恢复上下文
    }

    try {
        return point.proceed();
    } finally {
        if (firstArg instanceof BaseCommand || firstArg instanceof AbstractDomainEvent) {
            DomainEventContext.clear();
        }
    }
}

两件事：

注入 —— 命令/事件里的上下文搬进 ThreadLocal，聚合根和仓储随时能取。
清理 —— finally 里 DomainEventContext.clear()。线程池会复用线程，不清理就会把上一个请求的身份信息泄漏给下一个请求。这是必须有的，框架已经替你做了。

[6] CommandHandler —— 领域逻辑

你的代码。唯一的要求是：状态变更必须在聚合根里完成。

java

Producer domain = new Producer();
domain.create(command.getName(), command.getContact());   // 改状态 + 挂事件

AbstractAggregateRoot.addPendingEvent 在挂事件的同时就把上下文注入了：

java

protected void addPendingEvent(AbstractDomainEvent event) {
    if (event != null) {
        injectContext(event);      // ← 此刻 requestId / userName 就写进事件了
        domainEvents.add(event);
    }
}

所以即使事件永远没被发布，它身上也已经带着正确的上下文了。

[7] 仓储切面 —— 事件的触发点

java

@Around("execution(* *..*Repository+.save(..)) || " +
        "execution(* *..*Repository+.update(..)) || " +
        "execution(* *..*Repository+.delete(..)) || " +
        "execution(* *..*Repository+.operate(..))")
public Object aroundSave(ProceedingJoinPoint joinPoint) throws Throwable {
    Object result = joinPoint.proceed();     // ← 先真的保存

    Object[] args = joinPoint.getArgs();
    if (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()) {
                // 有事务 → 事务提交后再发
                TransactionSynchronizationManager.registerSynchronization(
                    new TransactionSynchronization() {
                        @Override
                        public void afterCommit() {
                            publishEvents(events);
                        }
                    });
            } else {
                publishEvents(events);       // 无事务 → 立即发
            }
        }
    }
    return result;
}

三个要点：

保存成功之后才发事件。 保存失败抛异常，事件不会发出去 —— 避免了"状态没存但事件发了"的经典不一致。
有事务时，事件在 afterCommit 才发。 事务回滚，事件也不会发。
clearDomainEvents() 在发布前调用。 事件只发一次，重复保存同一个聚合根不会重复发事件。

注意

切点的类型模式是 *Repository+：命中「名字以 Repository 结尾的类型」以及它的子类型。所以只要你实现的是 ProducerWriteRepository 这样的接口，实现类叫什么都会被拦到。反过来，如果接口和实现类的名字里都没有 Repository，切面就不认识它们 —— 保存成功了，但事件静默丢失，没有任何报错。

[8] 事件总线

EventBus 有两个实现，看 soda.event.bus-type 选：

实现	投递范围	特点
`SimpleEventBus`	进程内	默认，不依赖 Spring 事件机制
`RedisStreamEventBus`	跨服务	Redis Streams，带消费组、幂等、死信

提示

soda-event-spring 模块在 17.0.1 已经移除，没有 SpringEventBus。bus-type 只写 simple 或 redis。

注意

除了仓储切面，CqrsAroundHandler 还有一条发布路径：如果CommandHandler 的返回值实现了 DomainEvents（比如直接返回聚合根），或者返回值本身就是 DomainEvent，切面会在 send 返回后把还没被取走的事件直接发一次。正常写法是返回 DTO，让事件统一由仓储切面发布 —— 否则同一个事件可能发两遍。

提示

当 bus-type=redis 时，CqrsAroundHandler 会跳过本地立即发布，把发布权交给仓储切面 —— 避免同一条事件既进 Redis 又在本地跑一遍。

[9] 事件消费者 —— 上下文恢复

EventHandler 被调用时，切面从事件本身恢复上下文：

java

} else if (firstArg instanceof AbstractDomainEvent) {
    AbstractDomainEvent event = (AbstractDomainEvent) firstArg;
    DomainEventContext.setRequestId(event.getRequestId());
    DomainEventContext.setUserName(event.getUserName());
    DomainEventContext.setJti(event.getJti());
    DomainEventContext.setAuthorities(event.getAuthorities());
    DomainEventContext.setCallerUid(event.getCallerUid());
    DomainEventContext.setHopCount(event.getHopCount());
}

这就是上下文全链路传递的闭环：请求进来时从 HTTP 头拿，中途写进命令 → 事件，跨服务时序列化带着走，消费时再恢复回 ThreadLocal。

一张表看全局

环节	谁写的	上下文在哪	出错时
Controller	你的切面	命令对象字段	正常返回
命令总线	框架切面	回填进命令	`IllegalStateException`（递归超限）
CommandHandler	框架切面	ThreadLocal	抛给切面
聚合根	框架	注入进事件	—
仓储	框架切面	事件字段	保存失败则事件不发
事件总线	框架	事件字段	记日志，不影响主流程
事件消费者	框架切面	恢复进 ThreadLocal	记日志

下一步

上下文为什么能跨线程、跨服务 → 上下文传递
递归是怎么被掐断的 → 递归保护
切面的具体实现细节 → 事件总线与切面

请求是怎么跑完的#

全景#

[1] Controller —— 上下文的第一站#

[3] 命令总线切面 —— 上下文回填与 hopCount#

[5] Handler 切面 —— 上下文注入与清理#

[6] CommandHandler —— 领域逻辑#

[7] 仓储切面 —— 事件的触发点#

[8] 事件总线#

[9] 事件消费者 —— 上下文恢复#

一张表看全局#

下一步#

请求是怎么跑完的

全景

[1] Controller —— 上下文的第一站

[3] 命令总线切面 —— 上下文回填与 hopCount

[5] Handler 切面 —— 上下文注入与清理

[6] CommandHandler —— 领域逻辑

[7] 仓储切面 —— 事件的触发点

[8] 事件总线

[9] 事件消费者 —— 上下文恢复

一张表看全局

下一步