NodeJS应用中双PostgreSQL数据库的两阶段提交实现方案咨询

嘿，这个分布式事务的场景我之前在NodeJS项目里也处理过，不用自己从零开发事务管理器，基于pg-promise和PostgreSQL本身的特性就能搞定。下面给你详细的思路和实现方案：

核心思路：利用PostgreSQL的原生命令实现2PC #

PostgreSQL原生支持PREPARE TRANSACTION、COMMIT PREPARED和ROLLBACK PREPARED这几个命令，刚好能覆盖两阶段提交的核心流程。因为pg-promise每个db实例对应单个数据库连接，我们只需要创建两个独立的db实例，手动串联2PC的两个阶段即可。

具体实现步骤 #

1. 初始化两个数据库连接实例 #

首先分别创建指向两个PostgreSQL库的pg-promise实例：

const pgp = require('pg-promise')();

// 主数据库配置
const dbConfigPrimary = {
  host: 'localhost',
  port: 5432,
  database: 'primary_db',
  user: 'your_user',
  password: 'your_password'
};

// 冗余备份数据库配置
const dbConfigReplica = {
  host: 'localhost',
  port: 5433,
  database: 'replica_db',
  user: 'your_user',
  password: 'your_password'
};

const dbPrimary = pgp(dbConfigPrimary);
const dbReplica = pgp(dbConfigReplica);

2. 封装两阶段提交的通用函数 #

写一个通用函数封装完整的2PC流程，包含异常处理和回滚逻辑：

/**
 * 跨两个数据库执行两阶段提交
 * @param {string} sql1 - 主库写操作SQL
 * @param {Array} params1 - 主库SQL参数
 * @param {string} sql2 - 备份库写操作SQL
 * @param {Array} params2 - 备份库SQL参数
 * @returns {Promise<boolean>} 提交是否成功
 */
async function twoPhaseCommit(sql1, params1, sql2, params2) {
  // 生成全局唯一事务ID，用于标识跨库的同一事务
  const txId = `dist_tx_${Date.now()}_${Math.random().toString(36).slice(2, 10)}`;
  let txPrimary, txReplica;

  try {
    // 第一步：启动两个数据库的本地事务
    txPrimary = await dbPrimary.tx();
    txReplica = await dbReplica.tx();

    // 第二步：执行各自的写操作
    await txPrimary.none(sql1, params1);
    await txReplica.none(sql2, params2);

    // 第三步：准备阶段 - 通知两个库进入提交准备状态
    // 此步骤后数据库会锁定相关资源，等待最终提交/回滚指令
    await txPrimary.none('PREPARE TRANSACTION $1', txId);
    await txReplica.none('PREPARE TRANSACTION $1', txId);

    // 第四步：提交阶段 - 正式提交已准备的事务
    // 注意：不能再用之前的tx对象，需用全局db实例执行提交
    await dbPrimary.none('COMMIT PREPARED $1', txId);
    await dbReplica.none('COMMIT PREPARED $1', txId);

    console.log('跨库事务提交成功');
    return true;
  } catch (error) {
    console.error('跨库事务失败，开始回滚:', error.message);
    // 回滚逻辑：确保两个库都能回滚到一致状态
    try {
      // 先尝试回滚已进入准备状态的事务
      if (txPrimary) {
        await dbPrimary.none('ROLLBACK PREPARED $1', txId).catch(err => 
          console.warn('主库回滚准备事务失败，需人工检查:', err.message)
        );
      }
      if (txReplica) {
        await dbReplica.none('ROLLBACK PREPARED $1', txId).catch(err => 
          console.warn('备库回滚准备事务失败，需人工检查:', err.message)
        );
      }
      // 若未到准备阶段，直接回滚本地事务
      if (txPrimary && !txPrimary.isCompleted()) {
        await txPrimary.rollback();
      }
      if (txReplica && !txReplica.isCompleted()) {
        await txReplica.rollback();
      }
    } catch (rollbackErr) {
      console.error('回滚过程异常，需人工介入排查数据一致性:', rollbackErr.message);
    }
    // 抛出原始错误，让上层业务处理
    throw error;
  } finally {
    // 释放事务资源
    if (txPrimary) txPrimary.done();
    if (txReplica) txReplica.done();
  }
}

3. 在RESTful接口中调用 #

比如在Express路由里集成这个函数：

const express = require('express');
const app = express();
app.use(express.json());

app.post('/api/users', async (req, res) => {
  const { name } = req.body;
  const writeSqlPrimary = 'INSERT INTO users(name, created_at) VALUES($1, NOW())';
  const writeSqlReplica = 'INSERT INTO backup_users(name, created_at) VALUES($1, NOW())';

  try {
    await twoPhaseCommit(writeSqlPrimary, [name], writeSqlReplica, [name]);
    res.status(201).json({ message: '用户创建成功' });
  } catch (err) {
    res.status(500).json({ error: '用户创建失败，请重试' });
  }
});

app.listen(3000, () => console.log('服务启动在3000端口'));

关键注意事项 #

• 唯一事务ID：txId必须保证全局唯一，建议用时间戳+随机字符串组合，避免不同事务冲突。
• 极端场景处理：如果提交阶段一个库成功、另一个失败（如网络中断），会出现数据不一致，需人工介入检查修复，这是分布式系统无法完全避免的脑裂问题。
• 测试覆盖：一定要模拟各种失败场景（如准备阶段备库报错、提交阶段主库网络中断），验证回滚逻辑是否正常。
• 事务超时：PostgreSQL的准备事务不会自动超时，建议在业务层添加超时逻辑，避免长时间锁定资源。

内容的提问来源于stack exchange，提问作者The Once-ler