如何实现类似Prisma.js的透明事务系统？探究$transaction()实现方式

一、$transaction数组模式的实现思路 #

你提到的**检查上下文（this）**的思路其实是非常简洁可行的核心方案，Prisma的数组式事务本质就是复用现有模型方法，通过上下文绑定让这些方法在同一个事务连接中执行。

核心步骤 #

• 封装底层事务管理：基于数据库驱动的事务API（比如BEGIN/COMMIT/ROLLBACK语句，或驱动自带的transaction接口）实现事务的开启、提交与回滚。
• 创建事务专属上下文：调用$transaction时，生成一个绑定了事务连接的客户端实例，所有模型方法在这个上下文执行时，都会使用该事务连接而非默认连接池。
• 批量执行操作：遍历传入的操作数组，通过call()或apply()改变每个操作的执行上下文，让原本调用原客户端的方法转而使用事务上下文，最后统一提交或回滚。

简化实现示例 #

const { Pool } = require('pg');
const pool = new Pool({ /* 数据库配置 */ });

class MockPrisma {
  constructor(db) {
    this.db = db;
    this.post = new PostModel(this.db);
  }

  // 数组模式的$transaction
  async $transaction(operations) {
    let client;
    try {
      // 从连接池获取一个连接，用于事务
      client = await pool.connect();
      await client.query('BEGIN');
      
      // 创建绑定事务连接的客户端实例
      const txClient = new MockPrisma(client);
      
      // 执行所有操作，通过call切换上下文到txClient
      const results = await Promise.all(
        operations.map(op => op.call(txClient))
      );
      
      await client.query('COMMIT');
      return results;
    } catch (err) {
      if (client) await client.query('ROLLBACK');
      throw err;
    } finally {
      if (client) client.release();
    }
  }
}

class PostModel {
  constructor(db) {
    this.db = db;
  }

  async findMany({ where }) {
    const res = await this.db.query(
      'SELECT * FROM posts WHERE title LIKE $1',
      [`%${where.title.contains}%`]
    );
    return res.rows;
  }

  async count() {
    const res = await this.db.query('SELECT COUNT(*) FROM posts');
    return parseInt(res.rows[0].count, 10);
  }
}

// 使用示例
const prisma = new MockPrisma(pool);
const [posts, totalPosts] = await prisma.$transaction([
  prisma.post.findMany({ where: { title: { contains: 'prisma' } } }),
  prisma.post.count()
]);

二、透明事务系统的实现 #

Prisma的透明事务（即无需手动传递事务客户端，嵌套调用自动感知事务）核心依赖Node.js的Async Local Storage（ALS），它能在整个异步调用链中共享上下文，避免手动传递事务连接。

核心步骤 #

• 初始化Async Local Storage：用于在异步链中存储当前的事务连接。
• 改造客户端连接获取逻辑：所有模型方法优先从ALS中获取事务连接，没有则使用默认连接池。
• 事务回调封装：在$transaction的回调模式中，将事务连接存入ALS，回调内的所有异步操作都会自动复用该连接。

简化实现示例 #

const { Pool } = require('pg');
const { AsyncLocalStorage } = require('async_hooks');

const pool = new Pool({ /* 数据库配置 */ });
const als = new AsyncLocalStorage();

class MockPrisma {
  constructor() {
    this.post = new PostModel(this);
  }

  // 动态获取当前连接：优先用ALS中的事务连接，否则用连接池
  get db() {
    return als.getStore() || pool;
  }

  // 回调模式的透明事务
  async $transaction(callback) {
    let client;
    try {
      client = await pool.connect();
      await client.query('BEGIN');
      
      // 将事务连接存入ALS，回调内的所有异步操作都能获取到
      await als.run(client, async () => {
        // 回调内直接使用原prisma实例即可，无需传递tx
        await callback(this);
      });
      
      await client.query('COMMIT');
    } catch (err) {
      if (client) await client.query('ROLLBACK');
      throw err;
    } finally {
      if (client) client.release();
    }
  }
}

class PostModel {
  constructor(prisma) {
    this.prisma = prisma;
  }

  async findMany({ where }) {
    const res = await this.prisma.db.query(
      'SELECT * FROM posts WHERE title LIKE $1',
      [`%${where.title.contains}%`]
    );
    return res.rows;
  }

  async count() {
    const res = await this.prisma.db.query('SELECT COUNT(*) FROM posts');
    return parseInt(res.rows[0].count, 10);
  }
}

// 使用示例：回调内无需传递tx，自动感知事务
const prisma = new MockPrisma();
await prisma.$transaction(async () => {
  const posts = await prisma.post.findMany({ where: { title: { contains: 'prisma' } } });
  const totalPosts = await prisma.post.count();
  console.log(posts.length, totalPosts);
});

关键说明 #

• ALS确保了在异步调用链（包括嵌套函数、第三方库调用）中，所有模型方法都能拿到当前事务连接，实现了"透明"效果。
• 生产环境中还需要处理事务隔离级别、嵌套事务、连接泄漏防护等细节，但核心原理就是ALS+上下文感知的连接获取。

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