使用Knex.js与PostgreSQL处理并发事务的问题咨询

最终credits结果 #

按照你给出的执行序列，用户最终的credits会是**-10**，你的理解存在偏差。PostgreSQL的READ COMMITTED隔离级别下，每个事务内的查询基于事务首次读取时的快照，不会自动在提交前重新获取最新数据：

1. substractCredits事务读取到credits=10后，后续该事务内的所有查询都会沿用这个快照值；
2. addCredits提交后修改的30，substractCredits事务完全不可见；
3. substractCredits最终基于初始的10计算得到-10并提交，会直接覆盖掉addCredits设置的30。

解决并发冲突的方案 #

如果要避免这种覆盖问题，实现你期望的串行执行逻辑（addCredits等待substractCredits完成再操作），可采用以下几种方案：

1. 原子更新（最推荐） #

跳过应用层先查询再计算的步骤，直接让数据库完成原子性的增减操作，从根源避免并发问题。Knex提供了对应的便捷API：

// addCredits 改为原子递增
const addCredits = async (knex: Knex, userId: number) => {
  try {
    await knex('users')
      .where('id', userId)
      .increment('credits', 20);
  } catch (error) {
    // 处理错误逻辑
  }
}

// substractCredits 改为原子递减
const substractCredits = async (knex: Knex, userId: number) => {
  try {
    await knex('users')
      .where('id', userId)
      .decrement('credits', 20);
  } catch (error) {
    // 处理错误逻辑
  }
}

这种方式无需显式开启事务（Knex的increment/decrement本身是原子操作），数据库会保证每次操作都基于最新的credits值执行。

2. 悲观锁（FOR UPDATE） #

在查询用户数据时，给目标行加上排他锁，阻塞其他事务对该行的读写操作，直到当前事务完成。修改你的代码如下：

const addCredits = async (knex: Knex, userId: number) => {
  const trx = await knex.transaction();
  try {
    // 用forUpdate添加行级排他锁
    const user = await trx('users')
      .where('id', userId)
      .forUpdate()
      .first();
    user.credits += 20;
    await trx('users')
      .where('id', userId)
      .update({ credits: user.credits });
    await trx.commit();
  } catch (error) {
    await trx.rollback();
  }
}

const substractCredits = async (knex: Knex, userId: number) => {
  const trx = await knex.transaction();
  try {
    const user = await trx('users')
      .where('id', userId)
      .forUpdate()
      .first();
    user.credits -= 20;
    await trx('users')
      .where('id', userId)
      .update({ credits: user.credits });
    await trx.commit();
  } catch (error) {
    await trx.rollback();
  }
}

当substractCredits先查询加锁后，addCredits的查询会被阻塞，直到substractCredits提交或回滚，保证操作的串行性。

3. 乐观锁 #

给users表新增version字段（整数类型，默认0），每次更新时带上当前的version值，如果更新影响行数为0，说明数据已被其他事务修改，重试当前操作：

const addCredits = async (knex: Knex, userId: number) => {
  const trx = await knex.transaction();
  try {
    const user = await trx('users').where('id', userId).first();
    const updateResult = await trx('users')
      .where('id', userId)
      .where('version', user.version)
      .update({
        credits: user.credits + 20,
        version: user.version + 1
      });
    if (updateResult === 0) {
      // 更新失败，重试事务
      await trx.rollback();
      return addCredits(knex, userId);
    }
    await trx.commit();
  } catch (error) {
    await trx.rollback();
  }
}

乐观锁适合并发冲突较少的场景，不会阻塞其他事务，冲突时通过重试解决问题。

资金/积分场景通用方案 #

处理这类敏感数值操作，核心原则是避免在应用层读取数据后计算再写入，优先依赖数据库的原子能力：

• 优先使用数据库原生的原子增减语句（如PostgreSQL的UPDATE ... SET credits = credits + 20）；
• 若必须先读取数据做复杂逻辑，根据并发量选择乐观锁（低冲突场景）或悲观锁（高冲突场景）；
• 尽量缩短事务执行时间，减少锁的持有时长，避免阻塞其他操作；
• 禁止在事务内加入IO等待（如调用外部API），避免长事务；
• 跨服务并发场景可引入分布式锁（如Redis锁），单数据库场景下优先使用数据库自身的锁机制。

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