PSI 进销存系统分布式事务与数据一致性保障

发布时间：2026-06-21 | 分类：分布式架构

引言

在分布式架构下，进销存系统需要处理采购、销售、库存、财务等多个服务的数据一致性。本文将介绍 PSI 系统中的分布式事务实现方案，确保跨服务操作的原子性和一致性。

分布式事务方案对比

常见的分布式事务解决方案：

核心代码实现

TCC 事务管理器实现：

// TCC 事务管理器
class TCCTransactionManager {
  constructor(config) {
    this.transactionLog = config.transactionLog;
    this.participantRegistry = new Map();
    this.timeout = config.timeout || 30000;
  }

  // 注册参与者
  registerParticipant(transactionId, participant) {
    if (!this.participantRegistry.has(transactionId)) {
      this.participantRegistry.set(transactionId, []);
    }
    this.participantRegistry.get(transactionId).push(participant);
  }

  // 开启事务
  async beginTransaction(transactionId) {
    const transaction = {
      id: transactionId,
      status: 'pending',
      participants: [],
      startTime: Date.now(),
      createdAt: new Date()
    };

    await this.transactionLog.save(transaction);
    return transaction;
  }

  // 预 try
  async tryPhase(transactionId) {
    const transaction = await this.transactionLog.findById(transactionId);
    const participants = this.participantRegistry.get(transactionId) || [];

    const results = await Promise.allSettled(
      participants.map(p => p.try())
    );

    // 检查是否有失败
    const failures = results.filter(r => r.status === 'rejected');
    if (failures.length > 0) {
      // Try 阶段失败，进入取消阶段
      await this.cancelPhase(transactionId);
      throw new Error('Try phase failed, transaction rolled back');
    }

    // 更新事务状态
    transaction.status = 'try-success';
    await this.transactionLog.update(transaction);

    return transaction;
  }

  // 确认 commit
  async confirmPhase(transactionId) {
    const transaction = await this.transactionLog.findById(transactionId);
    const participants = this.participantRegistry.get(transactionId) || [];

    const results = await Promise.allSettled(
      participants.map(p => p.confirm())
    );

    const failures = results.filter(r => r.status === 'rejected');
    if (failures.length > 0) {
      // 记录确认失败，需要人工干预
      transaction.status = 'confirm-failed';
      transaction.error = failures.map(f => f.reason);
      await this.transactionLog.update(transaction);
      throw new Error('Confirm phase failed, requires manual intervention');
    }

    // 更新事务状态
    transaction.status = 'committed';
    transaction.commitTime = new Date();
    await this.transactionLog.update(transaction);

    // 清理注册信息
    this.participantRegistry.delete(transactionId);

    return transaction;
  }

  // 取消rollback
  async cancelPhase(transactionId) {
    const transaction = await this.transactionLog.findById(transactionId);
    const participants = this.participantRegistry.get(transactionId) || [];

    // 逆序取消
    for (let i = participants.length - 1; i >= 0; i--) {
      try {
        await participants[i].cancel();
      } catch (error) {
        // 记录取消失败
        console.error(`Participant ${i} cancel failed:`, error);
      }
    }

    // 更新事务状态
    transaction.status = 'cancelled';
    transaction.cancelTime = new Date();
    await this.transactionLog.update(transaction);

    return transaction;
  }

  // 执行事务
  async execute(transactionId, handler) {
    try {
      // 开启事务
      await this.beginTransaction(transactionId);

      // 执行业务逻辑
      await handler();

      // Try 阶段成功，执行确认
      await this.tryPhase(transactionId);

      // 发送确认指令
      await this.confirmPhase(transactionId);

      return { success: true, transactionId };

    } catch (error) {
      // 发生错误，取消事务
      await this.cancelPhase(transactionId);
      return { success: false, error: error.message };
    }
  }

  // 扫描超时事务
  async scanTimeoutTransactions() {
    const timeout = Date.now() - this.timeout;
    const transactions = await this.transactionLog.findTimeout(timeout);

    for (const tx of transactions) {
      if (tx.status === 'try-success') {
        // Try 阶段成功但未确认，重试确认
        await this.confirmPhase(tx.id).catch(console.error);
      } else if (tx.status === 'pending') {
        // Try 阶段超时，取消事务
        await this.cancelPhase(tx.id).catch(console.error);
      }
    }
  }
}

// TCC 参与者接口
class TCCParticipant {
  constructor(resourceId, service) {
    this.resourceId = resourceId;
    this.service = service;
    this.tryResult = null;
  }

  // Try 阶段：预留资源
  async try() {
    this.tryResult = await this.service.reserve(this.resourceId);
    return this.tryResult;
  }

  // Confirm 阶段：确认使用资源
  async confirm() {
    if (this.tryResult) {
      await this.service.confirm(this.resourceId, this.tryResult);
    }
  }

  // Cancel 阶段：释放预留的资源
  async cancel() {
    if (this.tryResult) {
      await this.service.cancel(this.resourceId, this.tryResult);
    }
  }
}

// 库存服务 TCC 实现
class InventoryTCCService {
  constructor(inventoryService) {
    this.inventoryService = inventoryService;
  }

  // Try: 扣减库存（预留）
  async reserve(params) {
    const { productId, warehouseId, quantity, orderId } = params;

    // 检查库存是否充足
    const available = await this.inventoryService.checkAvailable(
      productId, warehouseId, quantity
    );

    if (!available) {
      throw new Error('Insufficient inventory');
    }

    // 创建预留记录
    const reserveResult = await this.inventoryService.createReservation({
      productId,
      warehouseId,
      quantity,
      orderId,
      status: 'reserved',
      expireAt: new Date(Date.now() + 30 * 60 * 1000) // 30分钟过期
    });

    return reserveResult;
  }

  // Confirm: 确认使用预留
  async confirm(reserveId, reserveResult) {
    // 将预留转为正式库存扣减
    await this.inventoryService.confirmReservation(reserveId);

    // 扣减实际库存
    await this.inventoryService.decreaseStock({
      productId: reserveResult.productId,
      warehouseId: reserveResult.warehouseId,
      quantity: reserveResult.quantity,
      orderId: reserveResult.orderId,
      type: 'sale'
    });
  }

  // Cancel: 取消预留
  async cancel(reserveId, reserveResult) {
    // 释放预留
    await this.inventoryService.cancelReservation(reserveId);
  }
}

可靠消息方案

基于消息队列的最终一致性：

// 可靠消息服务
class ReliableMessageService {
  constructor(mqClient, db) {
    this.mqClient = mqClient;
    this.db = db;
    this.retryPolicy = {
      maxRetries: 3,
      retryInterval: 5000,
      exponentialBackoff: true
    };
  }

  // 发送消息
  async sendMessage(topic, message, options = {}) {
    const msg = {
      id: this.generateId(),
      topic,
      payload: JSON.stringify(message),
      status: 'pending',
      retryCount: 0,
      createdAt: new Date(),
      nextRetryAt: new Date(),
      ...options
    };

    // 存储消息
    await this.db.messageStore.save(msg);

    // 发送到消息队列
    try {
      await this.mqClient.send(topic, msg);

      // 更新发送状态
      msg.status = 'sent';
      msg.sentAt = new Date();
      await this.db.messageStore.update(msg);

      return msg.id;

    } catch (error) {
      msg.status = 'send-failed';
      msg.error = error.message;
      await this.db.messageStore.update(msg);
      throw error;
    }
  }

  // 发送事务消息
  async sendTransactionMessage(topic, message, transaction) {
    const msgId = this.generateId();

    // 注册事务回调
    transaction.on('commit', async () => {
      await this.sendMessage(topic, { ...message, msgId });
    });

    transaction.on('rollback', async () => {
      console.log(`Transaction rolled back, message ${msgId} not sent`);
    });

    return msgId;
  }

  // 消费消息
  async consumeMessage(topic, handler) {
    await this.mqClient.subscribe(topic, async (rawMessage) => {
      const message = JSON.parse(rawMessage);

      try {
        // 处理消息
        await handler(message);

        // 确认消息
        await this.ackMessage(message.id);

      } catch (error) {
        // 处理失败
        await this.handleMessageFailure(message, error);
      }
    });
  }

  // 确认消息
  async ackMessage(messageId) {
    const msg = await this.db.messageStore.findById(messageId);
    if (msg) {
      msg.status = 'acknowledged';
      msg.acknowledgedAt = new Date();
      await this.db.messageStore.update(msg);
    }
  }

  // 处理失败消息
  async handleMessageFailure(message, error) {
    const msg = await this.db.messageStore.findById(message.id);
    if (!msg) return;

    msg.retryCount++;
    msg.lastError = error.message;

    if (msg.retryCount >= this.retryPolicy.maxRetries) {
      // 超过最大重试次数，进入死信队列
      msg.status = 'dead-letter';
      msg.deadLetterAt = new Date();
      msg.deadLetterReason = error.message;
    } else {
      // 计算下次重试时间
      const interval = this.retryPolicy.exponentialBackoff
        ? this.retryPolicy.retryInterval * Math.pow(2, msg.retryCount - 1)
        : this.retryPolicy.retryInterval;

      msg.status = 'pending';
      msg.nextRetryAt = new Date(Date.now() + interval);
    }

    await this.db.messageStore.update(msg);
  }

  // 扫描并重发pending消息
  async resendPendingMessages() {
    const pending = await this.db.messageStore.findPending(
      new Date()
    );

    for (const msg of pending) {
      try {
        await this.mqClient.send(msg.topic, msg);
        msg.status = 'sent';
        msg.sentAt = new Date();
        await this.db.messageStore.update(msg);
      } catch (error) {
        msg.status = 'send-failed';
        msg.lastError = error.message;
        await this.db.messageStore.update(msg);
      }
    }
  }

  generateId() {
    return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
  }
}

// Saga 模式实现
class SagaOrchestrator {
  constructor(services, compensations) {
    this.services = services;
    this.compensations = compensations;
    this.executedSteps = [];
  }

  // 执行 Saga
  async execute(steps) {
    this.executedSteps = [];

    for (const step of steps) {
      try {
        // 执行步骤
        const result = await step.service(step.input);

        // 记录已执行的步骤
        this.executedSteps.push({
          step: step.name,
          result,
          compensation: step.compensation
        });

      } catch (error) {
        // 发生错误，执行补偿
        await this.compensate();
        throw new Error(`Saga failed at step ${step.name}: ${error.message}`);
      }
    }

    return this.executedSteps;
  }

  // 补偿
  async compensate() {
    // 逆序执行补偿操作
    for (let i = this.executedSteps.length - 1; i >= 0; i--) {
      const executed = this.executedSteps[i];

      try {
        await this.compensations[executed.step](executed.result);
        console.log(`Compensation for ${executed.step} succeeded`);
      } catch (error) {
        console.error(`Compensation for ${executed.step} failed:`, error);
        // 记录补偿失败，需要人工干预
      }
    }
  }
}

最佳实践建议

• 选择合适方案：根据业务一致性要求和性能需求选择事务方案
• 幂等设计：所有操作需支持幂等，避免重复执行导致数据错误
• 超时处理：设置合理的超时时间，避免长时间挂起
• 监控告警：监控事务状态，及时发现异常
• 人工兜底：关键场景预留人工干预入口

总结

PSI 进销存系统通过分布式事务技术保障数据一致性：

• TCC 模式：适用于库存扣减等需要预留资源的场景
• 可靠消息：适用于异步解耦、最终一致的场景
• Saga 模式：适用于长流程、多服务协作场景
• 综合使用：根据不同业务选择最合适的方案

通过合理选择和实现分布式事务方案，可以构建可靠的分布式进销存系统。