写一个命令

写链路是 soda 里最核心的一条链路。它比读链路多两件事：事务和领域事件。这一节把一条完整的写链路从零写出来。

我们用同一个例子贯穿全文：创建一个「生产车间」（Producer）聚合。

五个步骤

定义聚合根（继承 AbstractAggregateRoot），在业务方法里挂事件
定义 Command（继承 BaseCommand<R>）
实现 CommandHandler
定义写仓储接口（继承 WriteRepository<T>）并实现它
用组装器做领域对象 ↔ DO / DTO 的转换

第一步：聚合根

聚合根继承 AbstractAggregateRoot。业务方法负责改状态，并挂起领域事件：

java

package com.example.demo.domain.producer;

import com.example.demo.domain.enums.ProducerEventTypeEnum;
import com.example.demo.domain.producer.event.ProducerEvent;
import com.example.demo.utils.base.tools.BuKaUid;
import com.hibuka.soda.domain.aggregate.AbstractAggregateRoot;
import lombok.Data;

@Data
public class Producer extends AbstractAggregateRoot {
    /**
     * uid
     */
    private String uid;

    /**
     * 名称
     */
    private String name;

    public boolean create(String name) {
        this.uid = BuKaUid.genDomainUid();
        this.name = name;

        // 添加事件 - 使用生产者创建事件，而不是消费者创建事件
        addPendingEvent(new ProducerEvent(ProducerEventTypeEnum.CREATE_PRODUCER.getDesc(), this));
        return true;
    }
}

三个要点：

addPendingEvent 是 protected —— 外部代码没法往聚合里塞事件，事件只能由聚合自己产生。
它是挂起不是发布 —— 事件先放进 AbstractAggregateRoot 内部的列表，等到仓储 save 之后由切面发布。
挂起时自动注入上下文 —— requestId、userName、jti、authorities、callerUid 会从 DomainEventContext（ThreadLocal）里取出来补到事件上。

提示

事件类型的字符串来自 ProducerEventTypeEnum.CREATE_PRODUCER.getDesc()。枚举的写法见领域事件。

AbstractAggregateRoot 继承自 AbstractDomainEvent，所以聚合根对象上也有 eventId / eventType / occurredOn 字段。这是历史设计，业务上用不到。

第二步：定义 Command

Command 继承 BaseCommand<R>，R 是 CommandHandler 的返回类型：

java

package com.example.demo.application.cqrs.command;

import com.example.demo.adapter.inbound.web.dto.ProducerDTO;
import com.hibuka.soda.cqrs.command.BaseCommand;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;

@Data
public class CreateProducerCommand extends BaseCommand<ProducerDTO> {
    @Schema(description = "名称", example = "domain_test")
    private String name;
}

BaseCommand<R> 自带的上下文字段（都带 @Schema(hidden = true)，不出现在接口文档里）：

字段	类型	说明
`requestId`	`String`	链路追踪 ID
`callerUid`	`String`	调用方 UID
`userName`	`String`	用户名
`authorities`	`String`	权限串
`jti`	`String`	JWT token id
`jwt`	`String`	JWT token
`tenantId`	`String`	租户 ID
`userInfo`	`String`	用户信息
`extension`	`String`	扩展字段
`hopCount`	`Integer`	异步递归保护的跳数

重要

Command 里只放业务参数。上下文字段由框架填充，业务代码不要在 Command 里写业务规则 —— 那属于聚合根的职责。

第三步：实现 CommandHandler

CommandHandler<C extends Command<R>, R> 就是 CommandHandler 的接口，handle 签名是 R handle(C command) throws BaseException：

java

package com.example.demo.application.cqrs.handler.command;

import com.example.demo.application.cqrs.command.CreateProducerCommand;
import com.example.demo.adapter.inbound.web.dto.ProducerDTO;
import com.example.demo.adapter.inbound.web.assembler.ProducerWebAssembler;
import com.example.demo.domain.producer.Producer;
import com.example.demo.infrastructure.assembler.ProducerAssembler;
import com.example.demo.domain.producer.repository.ProducerWriteRepository;
import com.hibuka.soda.foundation.error.BaseException;
import com.hibuka.soda.foundation.io.ParamChecker;
import com.hibuka.soda.cqrs.command.CommandHandler;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;

@Component
@RequiredArgsConstructor
public class CreateProducerCommandHandler implements CommandHandler<CreateProducerCommand, ProducerDTO> {
    private final ProducerWriteRepository producerWriteRepository;

    @Override
    public ProducerDTO handle(CreateProducerCommand command) throws BaseException {
        commandValidate(command);
        Producer domain = ProducerAssembler.toDomain(command);
        domain.create(command.getName());
        producerWriteRepository.save(domain);
        return ProducerWebAssembler.toDTO(domain);
    }

    private void commandValidate(CreateProducerCommand command) throws BaseException {
        ParamChecker.notNull(command, "command");
        ParamChecker.notNull(command.getName(), "name");
    }
}

CommandHandler 的四步是固定的：

ParamChecker 校验入参
组装出聚合根
调聚合根的业务方法（改状态、挂事件）
存仓储，返回结果

注意

CommandHandler 里注入的必须是接口 ProducerWriteRepository，不能是实现类 ProducerWriteRepositoryImpl。否则你就在应用层直接依赖了基础设施层。

提示

Handler 不需要显式注册。总线在启动时扫描 List<CommandHandler<?, ?>>，用 GenericTypeResolver.resolveTypeArguments 解析出每个 CommandHandler 对应的命令类型并建索引。同一个命令类型只能有一个 CommandHandler，后注册的会覆盖前面的。三种 Handler 的完整对照见 Handler。

第四步：写仓储

接口

写仓储接口继承 WriteRepository<T extends AbstractAggregateRoot>：

java

package com.example.demo.domain.producer.repository;

import com.example.demo.domain.producer.Producer;
import com.hibuka.soda.foundation.error.BaseException;
import com.hibuka.soda.domain.repository.WriteRepository;

public interface ProducerWriteRepository extends WriteRepository<Producer> {
    /**
     * @param producer
     * @description: 保存Producer
     * @return: boolean
     */
    boolean save(Producer producer) throws BaseException;
}

WriteRepository<T> 固定四个方法：

java

public interface WriteRepository<T extends AbstractAggregateRoot> {
    boolean save(T aggregate) throws BaseException;
    boolean update(T aggregate) throws BaseException;
    boolean delete(T aggregate) throws BaseException;
    boolean operate(T aggregate, int operationType) throws BaseException;
}

operate 按操作类型分派，operationType 是 int，通常来自你自己的操作类型枚举。

实现

java

package com.example.demo.infrastructure.repositoryImpl.producer;

import com.example.demo.domain.producer.Producer;
import com.example.demo.domain.producer.repository.ProducerWriteRepository;
import com.example.demo.infrastructure.acl.mysql.entity.ScodaProducer;
import com.example.demo.infrastructure.acl.mysql.mapper.ScodaProducerMapper;
import com.example.demo.infrastructure.assembler.ProducerAssembler;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.hibuka.soda.foundation.error.BaseErrorCode;
import com.hibuka.soda.foundation.error.BaseException;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import com.example.demo.domain.enums.OperationTypeEnum;

@Component
@RequiredArgsConstructor
public class ProducerWriteRepositoryImpl implements ProducerWriteRepository {
    private final ScodaProducerMapper scodaProducerMapper;

    public boolean save(Producer producer) throws BaseException {
        ScodaProducer scodaProducer = ProducerAssembler.toDO(producer);
        try {
            return scodaProducerMapper.insert(scodaProducer) > 0;
        } catch (Exception e) {
            throw new BaseException(BaseErrorCode.DATABASE_ERROR.getCode(), e.getMessage());
        }
    }

    public boolean update(Producer producer) throws BaseException {
        try {
            LambdaUpdateWrapper<ScodaProducer> wrapper = new LambdaUpdateWrapper<>();
            wrapper.eq(ScodaProducer::getUid, producer.getUid())
                   .set(ScodaProducer::getName, producer.getName())
                   .set(ScodaProducer::getGmtModified, LocalDateTime.now());
            return scodaProducerMapper.update(null, wrapper) > 0;
        } catch (Exception e) {
            throw new BaseException(BaseErrorCode.DATABASE_ERROR.getCode(), e.getMessage());
        }
    }

    public boolean delete(Producer producer) throws BaseException {
        try {
            LambdaQueryWrapper<ScodaProducer> wrapper = new LambdaQueryWrapper<>();
            wrapper.eq(ScodaProducer::getUid, producer.getUid());
            return scodaProducerMapper.delete(wrapper) > 0;
        } catch (Exception e) {
            throw new BaseException(BaseErrorCode.DATABASE_ERROR.getCode(), e.getMessage());
        }
    }

    public boolean operate(Producer producer, int operationType) throws BaseException {
        if (operationType == OperationTypeEnum.CREATE.getCode()) {
            return save(producer);
        } else if (operationType == OperationTypeEnum.UPDATE.getCode()) {
            return update(producer);
        } else if (operationType == OperationTypeEnum.DELETE.getCode()) {
            return delete(producer);
        }
        throw new BaseException(BaseErrorCode.SYSTEM_ERROR.getCode(), "Unsupported operationType: " + operationType);
    }
}

四个方法名一个都不能改 —— soda 的 RepositoryEventAspect 就是按这四个名字做切点拦截的。改了名字，领域事件发不出去。

操作类型枚举：

java

@Getter
public enum OperationTypeEnum implements CodeEnum {
    CREATE(1, "create"),
    UPDATE(2, "update"),
    DELETE(3, "delete"),
    CUSTOM(99, "custom");

    private final Integer code;
    private final String description;

    OperationTypeEnum(Integer code, String description) {
        this.code = code;
        this.description = description;
    }

    @Override
    public String getDesc() {
        return this.description;
    }
}

注意

仓储类型要能匹配 *Repository。 切面的匹配表达式是 execution(* *..*Repository+.save(..))，+ 表示「这个名字的类型及其子类型」。所以接口叫 ProducerWriteRepository 就够了 —— 实现类叫 ProducerWriteRepositoryImpl 或者 ProducerRepoImpl 都会被拦到。但如果接口和实现类的名字里都没有 Repository（比如 ProducerStore + ProducerStoreImpl），切面就看不到它们：保存成功了，事件静默丢失，没有任何报错。

另外，任何以 Repository 结尾、且有 save 方法的类都会被这个切面看到。切面会检查第一个参数是不是 AbstractAggregateRoot 的子类，不是就跳过 —— 所以误伤有限。

第五步：组装器

数据层组装器负责 DO ↔ 领域对象，以及 Command → 领域对象：

java

package com.example.demo.infrastructure.assembler;

import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.example.demo.application.cqrs.command.CreateProducerCommand;
import com.example.demo.domain.producer.Producer;
import com.example.demo.infrastructure.acl.mysql.entity.ScodaProducer;

import java.time.LocalDateTime;
import java.util.stream.Collectors;

public class ProducerAssembler {

    public static Producer toDomain(ScodaProducer doRoot) {
        if (doRoot == null) {
            return null;
        }
        Producer producer = new Producer();
        producer.setUid(doRoot.getUid());
        producer.setName(doRoot.getName());
        return producer;
    }

    public static Page<Producer> toDomain(Page<ScodaProducer> doRoot) {
        if (doRoot == null) {
            return null;
        }
        Page<Producer> to = new Page<>();
        to.setTotal(doRoot.getTotal());
        to.setRecords(doRoot.getRecords().stream().map(i -> toDomain(i)).collect(Collectors.toList()));
        to.setCurrent(doRoot.getCurrent());
        to.setSize(doRoot.getSize());
        to.setPages(doRoot.getPages());
        return to;
    }

    public static Producer toDomain(CreateProducerCommand doRoot) {
        if (doRoot == null) {
            return null;
        }
        Producer producer = new Producer();
        producer.setName(doRoot.getName());
        return producer;
    }

    public static ScodaProducer toDO(Producer producer) {
        if (producer == null) {
            return null;
        }
        ScodaProducer to = new ScodaProducer();
        LocalDateTime currentTime = LocalDateTime.now();
        to.setUid(producer.getUid());
        to.setName(producer.getName());
        to.setGmtCreate(currentTime);
        to.setGmtModified(currentTime);
        return to;
    }
}

注意 toDomain(CreateProducerCommand) 只复制了 name，没有复制 uid —— 因为 uid 是聚合根在 create() 里生成的，业务代码不该自己造。

Web 层组装器负责领域对象 → DTO：

java

package com.example.demo.adapter.inbound.web.assembler;

public class ProducerWebAssembler {

    public static ProducerDTO toDTO(Producer producer) {
        if (producer == null) {
            return null;
        }
        ProducerDTO to = new ProducerDTO();
        to.setUid(producer.getUid());
        to.setName(producer.getName());
        return to;
    }
}

技巧

两个组装器分开放：数据层组装器（infrastructure.assembler）和 Web 层组装器（adapter.inbound.web.assembler）。DTO 是适配器层的东西，不该出现在基础设施层。

发命令：BusFacade

BusFacade 是业务代码接触总线的入口：

java

@Service
@RequiredArgsConstructor
public class SampleAppServiceImpl implements SampleAppService {
    private final BusFacade busFacade;

    @Override
    public ProducerDTO creatProducer(CreateProducerCommand command) throws BaseException {
        return busFacade.sendTransactCommand(command);
    }

    @Override
    public CompletableFuture<ProducerDTO> creatProducerSyn(CreateProducerCommand command) throws BaseException {
        return busFacade.sendAsyncCommand(command);
    }
}

四个发送方法：

方法	签名	说明
`sendCommand`	`<C extends Command<R>, R> R sendCommand(C)`	同步发送
`sendTransactCommand`	`<C extends Command<R>, R> R sendTransactCommand(C)`	事务内发送，写链路的默认选择
`sendAsyncCommand`	`<C extends Command<R>, R> CompletableFuture<R> sendAsyncCommand(C)`	异步发送，返回 `CompletableFuture`
`sendMqCommand`	`<C extends Command<R>, R> R sendMqCommand(C)`	未实现，调用会抛 `UnsupportedOperationException`
`sendQuery`	`<Q extends BaseQuery<R>, R> R sendQuery(Q)`	发查询

注意

sendMqCommand 当前是空实现，调用直接抛 UnsupportedOperationException("sendMqCommand feature is not yet implemented, do not call!")。不要用。

提示

sendCommand 和 sendTransactCommand 目前实现完全相同 —— 都是同步调 commandBus.send(command)。区别在于语义：sendTransactCommand 明确表达「这个命令要跑在事务里」，sendCommand 不承诺这一点。写链路请用 sendTransactCommand。

异步命令的上下文

sendAsyncCommand 会把当前的 CommandContext 捕获下来，在异步线程里恢复：

java

final CommandContext context = CommandContextHolder.getContext();
return CompletableFuture.supplyAsync(() -> {
    if (context != null) {
        CommandContextHolder.setContext(context);
    }
    try {
        return commandBus.send(command);
    } catch (BaseException e) {
        throw new RuntimeException(e);
    } finally {
        if (context != null) {
            CommandContextHolder.clearContext();
        }
    }
}, cqrsAsyncExecutor);

所以异步链路里 requestId、userName 这些上下文不会丢。

事务在哪

事务边界由你的应用服务或 Controller 上的 @Transactional 决定，soda 不替你开事务。

java

@Service
@RequiredArgsConstructor
public class ProducerAppServiceImpl implements ProducerAppService {
    private final BusFacade busFacade;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public ProducerDTO creatProducer(CreateProducerCommand command) throws BaseException {
        return busFacade.sendTransactCommand(command);
    }
}

rollbackFor = Exception.class 一定要写 —— 默认只对 RuntimeException 回滚，受检异常会提交事务。

重要

有没有事务，直接决定领域事件什么时候发布：

有事务 —— RepositoryEventAspect 注册 afterCommit 回调，事务提交成功后才发布事件。事务回滚，事件不发。
没事务 —— 事件在 save 返回后立即发布。

这就是「事务提交后才发事件」的实现方式。详见事件总线与切面。

完整链路一览

text

HTTP POST /api/creatProducer
  Header: Soda-Request-Id: abc-123
  Body:   {"name": "domain_test"}
  │
  ├─ ControllerAroundHandler
  │    ├─ requestId = "abc-123"，username 从请求头取
  │    ├─ command.setRequestId(...) / command.setUserName(...)
  │    └─ MDC.put("requestId", ...)
  │
  ├─ Controller：绑定 CreateProducerCommand，调应用服务
  │
  ├─ AppService（@Transactional）：busFacade.sendTransactCommand(command)
  │
  ├─ CommandBus.send(command)
  │    ├─ 按 Command 类型找到 CreateProducerCommandHandler
  │    └─ handler.handle(command)
  │
  ├─ CqrsAroundHandler（拦截 CommandBus.send）：记录耗时和入参出参
  │
  ├─ CreateProducerCommandHandler.handle(command)
  │    ├─ ParamChecker.notNull(command.getName(), "name")
  │    ├─ ProducerAssembler.toDomain(command) → Producer
  │    ├─ domain.create(name)
  │    │    ├─ uid = BuKaUid.genDomainUid()
  │    │    └─ addPendingEvent(new ProducerEvent("create_producer", this))  ← 挂起 + 注入上下文
  │    ├─ producerWriteRepository.save(domain)
  │    │    ├─ 切面拦截：真正执行 insert
  │    │    ├─ 取走 domain.getDomainEvents() 并清空
  │    │    └─ 注册 afterCommit 回调（事务提交后 publish）
  │    └─ ProducerWebAssembler.toDTO(domain) → ProducerDTO
  │
  │                     ↓ 事务提交
  │
  ├─ afterCommit → eventBus.publish(event)
  │    └─ ProducerEventHandler.handle(event)
  │
  └─ Result.success(dto)

常见问题

Q：命令没找到 Handler 会怎样？

SimpleCommandBus.send 抛 BaseException，错误码是 CLASS_NOT_FOUND_ERROR（10005），message 是 No handler registered for command: com.example.demo.application.cqrs.command.CreateProducerCommand。检查两件事：CommandHandler 有没有加 @Component，命令类的泛型有没有写对。

Q：能在一个 CommandHandler 里发另一个命令吗？

可以，但要注意递归。BusFacade 有同步递归深度保护：超过 10 层抛 IllegalStateException("Command recursion too deep! Trace: ...")。详见递归保护。

Q：更新和删除怎么写？

和创建一样，只是聚合根的业务方法不同（update / delete），仓储调 update / delete。RepositoryEventAspect 对这四个方法一视同仁 —— 都会在事务提交后发布事件。

Q：operate 必须实现吗？

必须。WriteRepository 声明了它，实现类不实现就编译不过。不想用就抛一个 BaseException 明确表示不支持。

Q：仓储方法返回 false 要处理吗？

建议处理。比如 save 返回 false 通常意味着影响行数为 0，这时事件仍然会被发布（切面在 save 返回后无条件取事件）。如果你希望「没保存成功就不发事件」，需要在 CommandHandler 里判断返回值并抛异常回滚。

下一步

读链路的写法：写一个查询
事件怎么发出去的：事件总线与切面
Web 层怎么接：接入 Web 层

写一个命令#

五个步骤#

第一步：聚合根#

第二步：定义 Command#

第三步：实现 CommandHandler#

第四步：写仓储#

接口#

实现#

第五步：组装器#

发命令：BusFacade#

异步命令的上下文#

事务在哪#

完整链路一览#

常见问题#

下一步#

写一个命令

五个步骤

第一步：聚合根

第二步：定义 Command

第三步：实现 CommandHandler

第四步：写仓储

接口

实现

第五步：组装器

发命令：BusFacade

异步命令的上下文

事务在哪

完整链路一览

常见问题

下一步