Axon 4 中处理外部 Partner 的 Commands 与 Events 设计

2026-10-24支付老李

结论

不要在 Payment Aggregate 的 Command Handler 中调用外部 Partner，也不必为每个 Event 创建一个 Command。

更合理的做法是把付款流程分成两个阶段：

RequestPaymentCommand 表达”请求付款”这一业务意图。Aggregate 完成校验后产生 PaymentRequestWasIssued。
Saga、Process Manager 或普通 Event Handler 监听该事件，并调用外部 Partner。
根据 Partner 返回的结果发送 AcceptPaymentCommand 或 RefusePaymentCommand，再由 Aggregate 产生 PaymentAccepted 或 PaymentRefused。

这些 Event 不应被强行放进同一个数据库事务。数据库事务无法可靠覆盖 HTTP、RPC 等外部调用。即使本地事务回滚，Partner 已执行的付款也不会自动撤销。这个流程需要最终一致性、幂等处理和失败重试，而不是试图建立跨系统的本地事务。

Command 与 Event 不是一一对应关系

Command 表达希望系统执行的业务动作，例如：

RequestPaymentCommand
AcceptPaymentCommand
RefusePaymentCommand

Event 表达已经发生且不可更改的事实，例如：

PaymentRequestWasIssued
PaymentAccepted
PaymentRefused

一个 Command 可能不产生 Event，也可能产生一个或多个 Event。同样，也不存在”每个 Event 必须有一个同名 Command”的要求。

“All the data that you need to apply the event should normally be available in the command”更准确的含义是：Command Handler 决定产生 Event 时，所需的业务输入应当明确可得。Aggregate 不应临时查询外部系统，并依赖不稳定的查询结果作出决定。这句话并不表示 Command 与 Event 必须一一对应。

为什么不应在 Aggregate 中调用 Partner

下面这种写法有明显风险：

@CommandHandler
public void handle(RequestPaymentCommand command) {
    apply(new PaymentRequestWasIssued(command.getPaymentId()));

    PartnerResult result = partnerClient.requestPayment(command);

    if (result.isAccepted()) {
        apply(new PaymentAccepted(command.getPaymentId()));
    } else {
        apply(new PaymentRefused(command.getPaymentId()));
    }
}

主要问题有：

Aggregate 的执行时间取决于 Partner 的响应速度。
Partner 超时会长时间占用 Unit of Work 和数据库资源。
Command 重试可能导致 Partner 被重复调用，甚至造成重复扣款。
Partner 调用成功后，如果本地 Event Store 提交失败，外部付款无法随本地事务回滚。
Aggregate 重放历史 Event 时不应触发外部副作用。
Aggregate 应专注于维护业务不变量，不应承担远程系统的编排工作。

即使 PaymentRequestWasIssued 和 PaymentAccepted 在 Axon 的同一个 Unit of Work 中提交，也只能保证本地事件存储的原子性，无法让 Partner 操作共享同一事务边界。

推荐的处理流程

1. Aggregate 接受付款请求

public record RequestPaymentCommand(
        String paymentId,
        BigDecimal amount,
        String currency
) {
}

Aggregate 负责检查金额、当前付款状态等业务规则：

@Aggregate
public class Payment {

    @AggregateIdentifier
    private String paymentId;

    private PaymentStatus status;

    protected Payment() {
    }

    @CommandHandler
    public Payment(RequestPaymentCommand command) {
        if (command.amount().signum() <= 0) {
            throw new IllegalArgumentException("Payment amount must be positive");
        }

        AggregateLifecycle.apply(new PaymentRequestWasIssued(
                command.paymentId(),
                command.amount(),
                command.currency()
        ));
    }

    @EventSourcingHandler
    public void on(PaymentRequestWasIssued event) {
        this.paymentId = event.paymentId();
        this.status = PaymentStatus.PENDING;
    }
}

此时，PENDING 是正常且明确的业务状态，不是事务尚未完成时的临时状态。

2. 在 Aggregate 外调用 Partner

这里可以使用 Event Handler。如果流程涉及超时、补偿、多个外部步骤或更复杂的状态协调，Axon Saga 会更合适。

@Component
public class PaymentPartnerProcess {

    private final PartnerClient partnerClient;
    private final CommandGateway commandGateway;

    public PaymentPartnerProcess(
            PartnerClient partnerClient,
            CommandGateway commandGateway
    ) {
        this.partnerClient = partnerClient;
        this.commandGateway = commandGateway;
    }

    @EventHandler
    public void on(PaymentRequestWasIssued event) {
        PartnerResult result = partnerClient.requestPayment(
                event.paymentId(),
                event.amount(),
                event.currency()
        );

        if (result.accepted()) {
            commandGateway.send(new AcceptPaymentCommand(
                    event.paymentId(),
                    result.partnerReference()
            ));
        } else {
            commandGateway.send(new RefusePaymentCommand(
                    event.paymentId(),
                    result.reasonCode()
            ));
        }
    }
}

Event Handler 重试时可能再次调用 Partner，因此请求中需要携带稳定的幂等键，例如 paymentId 或专门生成的 paymentRequestId。Partner 接口也应能通过该幂等键识别重复请求。

3. Aggregate 记录 Partner 的结果

public record AcceptPaymentCommand(
        String paymentId,
        String partnerReference
) {
}

public record RefusePaymentCommand(
        String paymentId,
        String reasonCode
) {
}

对应的 Command Handler：

@CommandHandler
public void handle(AcceptPaymentCommand command) {
    if (status == PaymentStatus.ACCEPTED) {
        return; // 幂等处理
    }

    if (status != PaymentStatus.PENDING) {
        throw new IllegalStateException(
                "Only a pending payment can be accepted"
        );
    }

    AggregateLifecycle.apply(new PaymentAccepted(
            command.paymentId(),
            command.partnerReference()
    ));
}

@CommandHandler
public void handle(RefusePaymentCommand command) {
    if (status == PaymentStatus.REFUSED) {
        return; // 幂等处理
    }

    if (status != PaymentStatus.PENDING) {
        throw new IllegalStateException(
                "Only a pending payment can be refused"
        );
    }

    AggregateLifecycle.apply(new PaymentRefused(
            command.paymentId(),
            command.reasonCode()
    ));
}

Event Sourcing Handler 负责更新 Aggregate 状态：

@EventSourcingHandler
public void on(PaymentAccepted event) {
    this.status = PaymentStatus.ACCEPTED;
}

@EventSourcingHandler
public void on(PaymentRefused event) {
    this.status = PaymentStatus.REFUSED;
}

第二个 Command 应如何命名

Command 的名称应体现业务意图，而不是照搬 Event 名称。通常有两种选择。

第一种是分别建模：

AcceptPaymentCommand
RefusePaymentCommand

这种命名的业务含义清楚，也允许不同结果使用各自的字段和校验规则。

第二种是使用统一的结果 Command：

public record CompletePaymentCommand(
        String paymentId,
        PaymentOutcome outcome,
        String partnerReference,
        String refusalReason
) {
}

如果两种结果的处理逻辑高度一致，可以采用这种方式。不过，它容易引入只在部分状态下使用的可空字段。因此，通常分别使用 AcceptPaymentCommand 和 RefusePaymentCommand 会更清楚。

如果想强调这些 Command 用于记录 Partner 已作出的决定，也可以命名为：

RecordPaymentAcceptedCommand
RecordPaymentRefusedCommand

但这些 Command 仍然需要经过 Aggregate 的业务校验，并非无条件写入 Event。

Partner 异步返回时的处理

如果 Partner 先接收请求，随后通过 webhook 或消息返回结果，流程可以改为：

RequestPaymentCommand
    -> PaymentRequestWasIssued
    -> 调用 Partner
    -> Partner 返回 requestId
    -> 等待 webhook 或消息
    -> AcceptPaymentCommand / RefusePaymentCommand
    -> PaymentAccepted / PaymentRefused

Partner 的回调入口应先完成鉴权、签名验证和数据格式转换，然后通过 Axon Command Gateway 发送 Command。HTTP Controller 不应直接修改 Aggregate 状态。

事务与可靠性的正确边界

每次本地状态变化应在各自的事务中原子提交：

RequestPaymentCommand 与 PaymentRequestWasIssued 属于一个本地事务。
AcceptPaymentCommand 与 PaymentAccepted 属于另一个本地事务。
RefusePaymentCommand 与 PaymentRefused 属于另一个本地事务。

跨系统的可靠性需要由以下机制保证：

为每次付款请求设置稳定且唯一的幂等键。
Partner 调用和回调都支持重复处理。
Event Handler 失败后能够安全重试。
明确建模超时状态，例如产生 PaymentTimedOut。
保存 Partner 的业务流水号，便于对账和排查。
为无法自动恢复的请求设置人工处理或补偿流程。
如果需要可靠发送消息，可以使用 transactional outbox，不要试图把消息中间件或 HTTP 调用放进数据库事务。
为 PENDING 状态设置超时机制。在 Axon 中，可以结合 Saga Deadline 或其他调度机制实现。

因此，PaymentRequestWasIssued 与最终结果 Event 分别属于不同事务并不是设计缺陷，而是对实际业务过程的准确描述：付款请求已经发出，但 Partner 尚未作出决定。这类跨系统流程应采用最终一致性模型。