如何用JavaScript实现一个支持容错处理的分布式事务？-js教程-PHP中文网

答案是采用Saga模式结合消息队列和幂等性设计实现分布式事务。通过将事务分解为本地操作序列，利用事件驱动机制推进流程，并在失败时执行补偿事务，确保最终一致性；编排器需持久化状态、保障消息可靠传递并处理超时与重试，以应对Node.js环境中的容错需求。

如何用javascript实现一个支持容错处理的分布式事务？

在JavaScript环境中实现一个支持容错的分布式事务，这本身就是个挑战，因为它不像传统关系型数据库那样有内置的ACID特性。在我看来，我们通常不是追求严格意义上的“分布式ACID事务”，而是在分布式系统中通过设计模式来达到“最终一致性”和“故障恢复能力”。最实际、也是我个人更倾向的方案，是采用Saga模式，尤其是结合消息队列和幂等性设计。

解决方案

要用JavaScript（通常是Node.js环境）实现一个支持容错的分布式事务，Saga模式是一个非常合适的选择。它将一个复杂的分布式事务分解成一系列本地事务，每个本地事务更新其所在服务的数据，并发布一个事件，触发下一个本地事务。如果某个本地事务失败，Saga会通过执行补偿事务来撤销之前成功的操作，从而确保系统最终达到一致状态。

我们以一个经典的订单创建流程为例：用户下单 -> 扣减库存 -> 处理支付 -> 安排发货。

启动Saga： 当用户请求创建订单时，一个“订单服务”会启动一个新的Saga实例。这个实例会保存Saga的当前状态和订单相关数据。
第一步：库存服务： 订单服务向一个可靠的消息队列（例如Kafka、RabbitMQ）发布一条消息，指示“库存服务”预留商品。
库存服务响应：
- 如果库存充足，库存服务预留商品，更新本地数据库，并发布一条“库存已预留”的成功事件。
- 如果库存不足，库存服务发布一条“库存预留失败”的事件。
Saga编排器（Orchestrator）决策： 一个专门的Saga编排器（可以是一个独立的Node.js服务，或者集成在订单服务中）会监听这些事件。
- 如果收到“库存已预留”的成功事件，编排器会更新Saga状态，并向“支付服务”发布一条消息，指示处理支付。
- 如果收到“库存预留失败”的事件，编排器会更新Saga状态为失败，并可能触发通知用户或重试机制。
第二步：支付服务： 支付服务收到消息后，处理支付。
- 支付成功，更新本地数据库，发布“支付成功”事件。
- 支付失败，发布“支付失败”事件。
编排器处理支付结果：
- 如果收到“支付成功”事件，编排器更新Saga状态，并向“物流服务”发布消息，安排发货。
- 如果收到“支付失败”事件，编排器更新Saga状态，并触发补偿事务：向“库存服务”发布消息，指示“释放之前预留的库存”。库存服务收到后执行补偿操作，并发布“库存已释放”事件。
第三步：物流服务： 物流服务收到消息后，安排发货。
- 发货成功，发布“发货成功”事件，Saga完成。
- 发货失败，发布“发货失败”事件，编排器触发补偿事务：向“支付服务”发布“退款”消息，向“库存服务”发布“释放库存”消息。

这个过程中，每个服务都只负责自己的本地事务，通过事件和消息队列进行通信，Saga编排器负责推进流程和处理故障时的补偿逻辑。

立即学习“Java免费学习笔记（深入）”；

为什么传统的两阶段提交（2PC）在现代JavaScript分布式系统中不适用？

在我看来，传统的两阶段提交（2PC）在理论上很严谨，但在现代、特别是基于Node.js的微服务架构中，它显得有些格格不入，甚至可以说是不切实际。我个人在实践中几乎没有看到有人在Node.js微服务场景中真正落地2PC。

主要原因有几个：

2PC的核心思想是有一个协调者（Coordinator）来协调所有参与者（Participants）的事务。它分为“准备阶段”和“提交阶段”。在准备阶段，协调者询问所有参与者是否准备好提交，参与者会锁定资源并回应。只有所有参与者都回应“是”，协调者才会在提交阶段指示所有参与者提交。任何一个参与者说“不”，或者超时未响应，协调者都会指示所有参与者回滚。

这听起来很美好，但问题在于：

同步阻塞： 2PC本质上是一个同步阻塞协议。在准备阶段，所有参与者都必须锁定它们涉及的资源，等待协调者的指令。这在高性能、高并发的Node.js服务中是致命的。Node.js的优势在于其非阻塞I/O和事件驱动模型，而2PC的阻塞特性会严重影响系统的吞吐量和响应时间。想象一下，一个服务为了等待其他服务响应而长时间持有数据库锁，这无疑会成为整个系统的瓶颈。
单点故障： 协调者是2PC的单点。如果协调者在提交阶段之前崩溃，参与者可能会一直处于资源锁定的“不确定”状态，这需要复杂的手动干预来解决，或者依赖非常复杂的恢复机制。这与分布式系统追求的高可用性原则相悖。
网络延迟与超时： 在分布式环境中，网络是不可靠的。多次跨网络通信（协调者与每个参与者之间的准备、提交/回滚）增加了延迟和超时的可能性。任何一个参与者的网络问题都可能导致整个事务失败或挂起。
数据库耦合： 2PC通常需要底层数据库支持XA事务（eXtended Architecture），这在异构微服务架构中很难实现。不同的服务可能使用不同的数据库技术（例如MongoDB、PostgreSQL、Redis），它们之间很难通过一个统一的XA接口来协调事务。Node.js服务更倾向于使用NoSQL或非XA兼容的数据库，这使得2PC的实现变得异常困难。

所以，在我看来，与其尝试在Node.js中硬性实现一个笨重且脆弱的2PC，不如拥抱分布式系统本身的特性，接受最终一致性，并通过Saga这样的模式来管理复杂业务流程中的数据一致性和故障恢复。

挖错网

一款支持文本、图片、视频纠错和AIGC检测的内容审核校对平台。

查看详情

如何设计一个健壮的Saga编排器（Orchestrator）来处理故障？

设计一个健壮的Saga编排器是实现容错分布式事务的关键。这不仅仅是写几行代码那么简单，它需要深入思考状态管理、通信机制以及各种异常场景的处理。

Saga状态的持久化： 这是编排器能够容错的基础。编排器自身也可能崩溃，所以它必须能够从上次成功的状态恢复。每次Saga状态发生变化（例如，从
```
INITIATED
```
登录后复制
到
```
INVENTORY_RESERVED
```
登录后复制
），或者编排器发送了命令给参与者服务后，都应该将Saga的当前状态以及所有相关上下文数据（如订单ID、商品列表、支付金额等）持久化到数据库中。这样，即使编排器重启，也能从数据库中加载Saga实例，并从中断的地方继续执行。
- 实现细节： 可以创建一个专门的
```
SagaLog
```
  登录后复制
  或
```
SagaState
```
  登录后复制
  表，存储
```
sagaId
```
  登录后复制
  、
```
currentStatus
```
  登录后复制
  、
```
payload
```
  登录后复制
  （JSON格式存储上下文数据）、
```
lastUpdated
```
  登录后复制
  等字段。
可靠的消息通信： 编排器与参与者服务之间的通信必须是可靠的。这意味着消息不能丢失，也不能重复处理。
- 消息队列（Message Queue）： 使用像Kafka、RabbitMQ这样的消息队列是最佳实践。它们提供了消息的持久化、At-Least-Once Delivery（至少一次投递）和At-Most-Once Delivery（至多一次投递）保证。
- 幂等性（Idempotency）： 参与者服务接收到消息后，执行的本地事务必须是幂等的。这意味着即使同一条消息被重复投递和处理多次，结果也应该是一致的，不会产生副作用。例如，预留库存操作可以检查是否已经预留过，支付操作可以检查是否已经支付过。
- Outbox Pattern（发件箱模式）： 这是一个非常重要的模式，用于确保本地数据库事务和消息发布是原子性的。当编排器更新Saga状态并需要发布消息时，它首先将消息写入自己的本地数据库的“发件箱”表，作为本地事务的一部分。然后，一个独立的“消息转发器”进程会定期扫描发件箱表，将消息发布到消息队列，并在成功发布后从发件箱中删除。这保证了Saga状态的更新和消息的发布要么都成功，要么都失败。
超时与重试机制：
- 命令超时： 编排器发送命令给参与者服务后，应该设置一个合理的超时时间。如果在超时时间内没有收到参与者服务的响应事件，编排器应该假定该命令失败，并触发重试（如果该操作可重试）或补偿流程。
- 重试策略： 对于瞬时故障，可以采用指数退避（Exponential Backoff）策略进行重试。但要注意，无限重试可能导致资源耗尽，需要设定最大重试次数。
- 死信队列（Dead-Letter Queue）： 对于那些经过多次重试仍然失败的消息，可以将其发送到死信队列，以便后续人工审查和处理，避免消息丢失。
补偿逻辑的健壮性： 补偿事务是Saga模式容错的核心。
- 明确定义补偿操作： 每个正向操作都必须有一个明确定义的、能够撤销其效果的补偿操作。例如，“预留库存”的补偿是“释放库存”，“处理支付”的补偿是“退款”。
- 补偿操作的幂等性： 补偿操作本身也必须是幂等的，以防补偿消息被重复发送。
- 补偿失败的处理： 如果补偿操作本身也失败了怎么办？这是一个更深层次的问题。通常，对于关键的补偿失败，需要有警报机制，并可能需要人工介入。有时，可以将失败的补偿请求放入一个“待处理补偿队列”，等待系统恢复或人工干预后再次尝试。

状态机设计： 将Saga编排器视为一个状态机，定义清晰的状态转换规则。每个状态都应该有明确的入口和出口条件，以及对应的处理逻辑。这有助于理清复杂的业务流程和故障路径。

// 概念性Saga状态机示例
const sagaStates = {
    INITIATED: 'initiated',
    INVENTORY_RESERVED: 'inventory_reserved',
    PAYMENT_PROCESSED: 'payment_processed',
    SHIPPING_SCHEDULED: 'shipping_scheduled',
    // 补偿状态
    COMPENSATING_INVENTORY: 'compensating_inventory',
    COMPENSATING_PAYMENT: 'compensating_payment',
    // 最终状态
    COMPLETED: 'completed',
    FAILED: 'failed'
};

class OrderSagaOrchestrator {
    constructor(messageBroker, sagaRepository) {
        this.messageBroker = messageBroker;
        this.sagaRepository = sagaRepository; // 负责Saga状态的持久化
        // 状态处理器映射
        this.stateHandlers = {
            // ... 其他状态
        };
    }

    async processEvent(event) {
        const { sagaId, type, payload } = event;
        let saga = await this.sagaRepository.findById(sagaId);

        if (!saga) {
            // 可能是首次事件，或者Saga已被清理，需要根据业务逻辑处理
            console.warn(`Saga ${sagaId} not found for event ${type}.`);
            return;
        }

        const handler = this.stateHandlers[saga.status];
        if (handler) {
            await handler.call(this, saga, payload); // 执行状态对应的处理逻辑
            await this.sagaRepository.save(saga); // 持久化Saga状态
        } else {
            console.error(`No handler for saga ${sagaId} in status ${saga.status}`);
        }
    }

    async handleInitiated(saga, payload) {
        // 收到来自库存服务的事件
        if (payload.status === 'inventory_reserved_success') {
            saga.status = sagaStates.INVENTORY_RESERVED;
            await this.messageBroker.publish('payment_service_topic', { type: 'process_payment', sagaId: saga.id, ...payload });
        } else if (payload.status === 'inventory_reserved_failed') {
            saga.status = sagaStates.FAILED;
            // 无需补偿，因为库存服务没有成功操作
        }
    }

    async handleInventoryReserved(saga, payload) {
        // 收到来自支付服务的事件
        if (payload.status === 'payment_success') {
            saga.status = sagaStates.PAYMENT_PROCESSED;
            await this.messageBroker.publish('shipping_service_topic', { type: 'schedule_shipping', sagaId: saga.id, ...payload });
        } else if (payload.status === 'payment_failed') {
            saga.status = sagaStates.COMPENSATING_INVENTORY;
            await this.messageBroker.publish('inventory_service_topic', { type: 'release_inventory', sagaId: saga.id, ...payload });
        }
    }

    // ... 其他状态处理函数，包括所有补偿逻辑
}

登录后复制

设计一个健壮的Saga编排器，核心在于“有状态”和“可靠通信”，并为各种失败场景预设好回滚路径。这需要细致的规划和大量的测试。