基于NestJS+Prisma+MySQL的支付网关并发提现请求处理问题

问题描述 #

我正在开发基于NestJS、Prisma ORM和MySQL的支付网关系统，遇到了并发提现请求的余额计算错误问题：当多个提现请求同时发起时，系统会读取到相同的用户可用余额，导致最终扣除后的余额与实际应剩金额不符。

具体场景：每个请求都会从用户可用余额中扣除提现金额，但并发请求会获取到相同的初始余额值——比如用户余额100，两个各提50的请求同时进来，都读到100，最后余额变成50而不是0。

已尝试方案 #

我试过用SELECT ... FOR UPDATE锁定用户行，并且用Prisma的$transaction包裹整个提现流程来保证原子性，相关代码如下：

async initiatePayout(payload: PayoutInitiateRequestDto, req: IRequest): Promise<PayoutResponse> {
  return await this.prisma.$transaction(async (prisma) => {
    // 锁定用户行，用于当前事务
    const user = await prisma.$queryRaw<User[]>`SELECT * FROM "User" WHERE "id" = ${userId} FOR UPDATE`;

    if (!user.length) {
      throw new Error('用户不存在');
    }

    const availableBalance = user[0]?.totalPayout;
    const payoutAmount = +payload.amount;

    // 检查余额是否充足
    if (availableBalance < payoutAmount) {
      throw new Error('余额不足');
    }

    // 调用外部支付API处理提现
    const apiResponse = await this.externalApi.payout(payload);

    if (apiResponse.success) {
      // 提现成功后扣除余额
      await this.user.findOneAndUpdate(
        { id: userId },
        { totalPayout: availableBalance - payoutAmount }
      );
    }

    return apiResponse;
  });
}

问题分析与修复建议 #

1. 事务上下文不一致：你在事务里用prisma.$queryRaw锁定了行，但更新余额时用的是this.user.findOneAndUpdate——这个操作没有使用事务回调内的prisma实例，会脱离当前事务上下文，导致行锁的作用失效。必须统一使用事务内的tx（即回调参数里的prisma）执行所有数据库操作。

2. 余额更新逻辑存在风险：用availableBalance - payoutAmount的方式更新余额，依赖于之前读取的内存值，即便在事务内，也应该用数据库端的原子更新（比如totalPayout = totalPayout - ${payoutAmount}），彻底避免变量计算带来的误差。

3. 外部API调用的事务一致性问题：外部API调用属于事务外操作，如果API调用成功但数据库更新失败，会出现状态不一致。建议调整流程：

   • 先执行数据库预扣操作；
   • 再调用外部API，若API失败则触发事务回滚；
   • 增加提现记录的状态字段（如pending/success/failed），通过定时任务处理异常状态的请求，保证最终一致性。

修复后的示例代码 #

async initiatePayout(payload: PayoutInitiateRequestDto, req: IRequest): Promise<PayoutResponse> {
  const userId = req.user.id;
  const payoutAmount = +payload.amount;

  return await this.prisma.$transaction(async (tx) => {
    // 锁定用户行并获取当前余额，使用事务内的tx实例
    const user = await tx.user.findUnique({
      where: { id: userId },
      select: { totalPayout: true }
    });

    if (!user) {
      throw new Error('用户不存在');
    }

    if (user.totalPayout < payoutAmount) {
      throw new Error('余额不足');
    }

    // 原子预扣余额
    await tx.user.update({
      where: { id: userId },
      data: { totalPayout: { decrement: payoutAmount } }
    });

    try {
      // 调用外部提现API
      const apiResponse = await this.externalApi.payout(payload);

      if (!apiResponse.success) {
        throw new Error('外部提现失败');
      }

      // 记录成功提现记录
      await tx.payoutRecord.create({
        data: {
          userId,
          amount: payoutAmount,
          status: 'SUCCESS',
          externalTransactionId: apiResponse.transactionId
        }
      });

      return apiResponse;
    } catch (error) {
      // 异常触发事务回滚，同时记录失败记录
      await tx.payoutRecord.create({
        data: {
          userId,
          amount: payoutAmount,
          status: 'FAILED',
          errorMessage: error.message
        }
      });
      throw error;
    }
  }, {
    // 可选：设置事务超时时间
    timeout: 5000
  });
}

额外注意事项 #

• 确保MySQL隔离级别为默认的REPEATABLE READ，配合行锁可有效防止不可重复读和幻读；
• 对提现请求做幂等性处理，比如通过请求ID或外部交易ID避免重复处理；
• 监控事务锁等待情况，避免长时间锁占用导致系统性能下降。

内容的提问来源于stack exchange，提问作者Utkarsh Sinha