Node.js 中的事务管理：Unit of Work 模式在服务层的应用

大家好，今天我们来深入探讨一个在现代后端开发中非常关键但又常常被忽视的话题——事务管理。特别是在使用 Node.js 这类非阻塞、事件驱动架构时，如何优雅地处理多个数据库操作之间的原子性问题，是一个必须掌握的核心技能。

我们将聚焦于 Unit of Work（工作单元）模式，这是一种经典的设计模式，尤其适用于复杂业务逻辑涉及多表写入或读取的场景。它不仅能帮助我们更好地组织代码结构，还能显著提升事务控制的可维护性和可靠性。

---

一、什么是 Unit of Work 模式？

定义

Unit of Work 是一种设计模式，用于跟踪一组对象的变化，并将这些变化作为一个整体提交到持久化存储（如数据库）。
它确保所有相关操作要么全部成功，要么全部失败回滚。

这个模式最早出现在 .NET 的 Entity Framework 和 Java 的 Hibernate 中，但在 Node.js 中同样适用，尤其是在结合 ORM（如 Sequelize、TypeORM）或原生 SQL 查询时。

核心思想

• 把多个数据库操作封装成一个“工作单元”。
• 在执行前记录所有变更（insert/update/delete）。
• 提交时统一应用；若出错则全部回滚。
• 避免手动管理每个查询的 begin/commit/rollback。

这与传统做法形成鲜明对比：

传统方式	Unit of Work
手动调用 BEGIN、COMMIT、ROLLBACK	自动管理事务生命周期
易漏掉 rollback 导致脏数据	强制一致性保障
逻辑分散，难以测试	统一入口，便于单元测试

---

二、为什么需要 Unit of Work？举个例子说明

假设你正在构建一个电商系统中的订单创建功能：

// 伪代码示例：不使用 Unit of Work 的风险
async function createOrder(userId, items) {
  const user = await db.query('SELECT * FROM users WHERE id = ?', [userId]);
  if (!user) throw new Error('User not found');

  const order = await db.query(
    'INSERT INTO orders (user_id, total) VALUES (?, ?)',
    [userId, calculateTotal(items)]
  );

  for (const item of items) {
    await db.query(
      'INSERT INTO order_items (order_id, product_id, quantity) VALUES (?, ?, ?)',
      [order.id, item.productId, item.quantity]
    );
    await db.query(
      'UPDATE inventory SET stock = stock - ? WHERE product_id = ?',
      [item.quantity, item.productId]
    );
  }
}

这段代码看似合理，但如果中间某个 UPDATE inventory 出现异常（比如库存不足），整个流程会中断，而前面已经插入了订单和部分订单项，造成数据不一致！

这就是典型的“部分提交”问题 —— 数据库状态处于中间态。

✅ 正确的做法是把这一整套操作放在同一个事务里，要么全成功，要么全失败。

---

三、实现 Unit of Work 模式（基于 PostgreSQL + Node.js）

我们将以 PostgreSQL 为例，使用 pg 或 sequelize 库作为底层驱动。这里我们选择更灵活的方式：直接使用 pg 并封装事务逻辑。

步骤 1：定义基础事务类（Unit of Work）

// unitOfWork.js
const { Pool } = require('pg');

class UnitOfWork {
  constructor(pool) {
    this.pool = pool;
    this.operations = []; // 存储待执行的操作
    this.isInTransaction = false;
  }

  async begin() {
    if (this.isInTransaction) throw new Error('Already in transaction');

    const client = await this.pool.connect();
    await client.query('BEGIN');

    this.client = client;
    this.isInTransaction = true;
    return this;
  }

  async commit() {
    if (!this.isInTransaction) throw new Error('Not in transaction');

    try {
      await this.client.query('COMMIT');
      this.isInTransaction = false;
      this.client.release();
    } catch (err) {
      await this.client.query('ROLLBACK');
      this.isInTransaction = false;
      this.client.release();
      throw err;
    }
  }

  async rollback() {
    if (!this.isInTransaction) throw new Error('Not in transaction');

    await this.client.query('ROLLBACK');
    this.isInTransaction = false;
    this.client.release();
  }

  // 添加操作到队列（延迟执行）
  addOperation(operation) {
    this.operations.push(operation);
  }

  // 执行所有操作（通常在 commit 前调用）
  async executeOperations() {
    for (const op of this.operations) {
      await op.execute(this.client);
    }
  }
}

步骤 2：定义具体操作类（Command Pattern）

为了让 Unit of Work 更易扩展，我们可以为每种数据库操作创建一个“命令对象”。

// commands/orderCommands.js
class InsertOrderCommand {
  constructor(orderData) {
    this.orderData = orderData;
  }

  async execute(client) {
    const result = await client.query(
      'INSERT INTO orders (user_id, total) VALUES ($1, $2) RETURNING id',
      [this.orderData.userId, this.orderData.total]
    );
    return result.rows[0].id;
  }
}

class InsertOrderItemCommand {
  constructor(orderId, productId, quantity) {
    this.orderId = orderId;
    this.productId = productId;
    this.quantity = quantity;
  }

  async execute(client) {
    await client.query(
      'INSERT INTO order_items (order_id, product_id, quantity) VALUES ($1, $2, $3)',
      [this.orderId, this.productId, this.quantity]
    );
  }
}

class UpdateInventoryCommand {
  constructor(productId, quantity) {
    this.productId = productId;
    this.quantity = quantity;
  }

  async execute(client) {
    const result = await client.query(
      'UPDATE inventory SET stock = stock - $1 WHERE product_id = $2 AND stock >= $1 RETURNING stock',
      [this.quantity, this.productId]
    );

    if (result.rows.length === 0) {
      throw new Error(`Insufficient stock for product ${this.productId}`);
    }
  }
}

步骤 3：服务层整合 Unit of Work（核心！）

现在我们把这些组件组合起来，在服务层完成真正的业务逻辑封装：

// services/orderService.js
const { Pool } = require('pg');
const UnitOfWork = require('../unitOfWork');
const { InsertOrderCommand, InsertOrderItemCommand, UpdateInventoryCommand } = require('../commands/orderCommands');

const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
});

class OrderService {
  async createOrder(userId, items) {
    const unitOfWork = new UnitOfWork(pool);

    try {
      await unitOfWork.begin();

      // Step 1: 创建订单
      const orderCmd = new InsertOrderCommand({ userId, total: this.calculateTotal(items) });
      const orderId = await orderCmd.execute(unitOfWork.client);

      // Step 2: 插入订单项
      for (const item of items) {
        const insertItemCmd = new InsertOrderItemCommand(orderId, item.productId, item.quantity);
        unitOfWork.addOperation(insertItemCmd);
      }

      // Step 3: 更新库存（批量检查）
      for (const item of items) {
        const updateInvCmd = new UpdateInventoryCommand(item.productId, item.quantity);
        unitOfWork.addOperation(updateInvCmd);
      }

      // 执行所有操作
      await unitOfWork.executeOperations();

      await unitOfWork.commit();
      return { success: true, orderId };

    } catch (error) {
      await unitOfWork.rollback();
      throw error; // 可以记录日志后再抛出
    }
  }

  calculateTotal(items) {
    return items.reduce((sum, item) => sum + item.price * item.quantity, 0);
  }
}

module.exports = new OrderService();

这样做的好处是什么？

特性	说明
✅ 原子性保证	所有操作要么一起成功，要么一起回滚
✅ 错误隔离	单个步骤失败不会污染其他已执行的部分
✅ 易于扩展	新增操作只需新增 Command 类即可
✅ 测试友好	可以 mock client 实现单元测试
✅ 日志清晰	回滚时能知道哪些操作被执行过

---

四、高级技巧：嵌套事务 & 多个 Unit of Work

有时候你会遇到这样的需求：在一个大事务中，需要拆分成几个子事务，或者跨多个模块共享事务上下文。

例如，用户注册时可能同时更新用户信息、发送欢迎邮件、创建默认配置等 —— 如果其中一个失败，整体应该回滚。

这时可以引入 嵌套事务（Savepoints）：

// 支持 savepoint 的版本
class UnitOfWorkWithSavepoints extends UnitOfWork {
  async savepoint(name) {
    await this.client.query(`SAVEPOINT ${name}`);
  }

  async releaseSavepoint(name) {
    await this.client.query(`RELEASE SAVEPOINT ${name}`);
  }

  async rollbackToSavepoint(name) {
    await this.client.query(`ROLLBACK TO SAVEPOINT ${name}`);
  }
}

然后在服务中使用：

async createAccount(userData) {
  const uow = new UnitOfWorkWithSavepoints(pool);
  await uow.begin();

  try {
    const userId = await this.createUser(uow, userData);

    // 子事务：创建用户配置
    await uow.savepoint('config');
    await this.createDefaultConfig(uow, userId);

    // 子事务：发送邮件（如果失败只回滚到 config）
    await uow.savepoint('email');
    await this.sendWelcomeEmail(userId);

    await uow.commit();
  } catch (err) {
    if (err.message.includes('email')) {
      await uow.rollbackToSavepoint('email'); // 回滚到 email 点
    } else if (err.message.includes('config')) {
      await uow.rollbackToSavepoint('config'); // 回滚到 config 点
    } else {
      await uow.rollback(); // 全部回滚
    }
    throw err;
  }
}

这种方式非常适合微服务间协作或复杂业务流程中的局部错误恢复。

---

五、常见陷阱与最佳实践总结

陷阱	解决方案
忘记调用 commit() 或 rollback()	使用 try-finally 或 async/await 包裹确保释放资源
多次连接池冲突	使用单一连接池实例，避免并发访问同一客户端
不合理的事务粒度	小事务优先（如单个 CRUD），避免长时间持有锁
缺乏监控	加入日志记录（如 Winston）追踪事务开始/结束时间
单元测试困难	使用内存数据库（如 SQLite）模拟事务行为进行测试

最佳实践建议：

1. 永远不要裸奔事务：始终封装在 Unit of Work 中。
2. 事务越短越好：减少锁竞争，提高并发性能。
3. 日志先行：记录事务 ID、操作类型、耗时，方便排查问题。
4. 异步事务要小心：Node.js 的 event loop 可能导致意外行为，推荐同步链式调用。
5. 考虑使用 ORM 的内置事务支持（如 Sequelize 的 transaction API），但理解其底层机制仍很重要。

---

六、结语：Unit of Work 是现代 Node.js 后端的必备武器

通过本文的学习，你应该已经掌握了：

• Unit of Work 的本质：将多个数据库操作视为一个逻辑单元
• 如何在 Node.js 中实现一个轻量级但健壮的事务管理器
• 如何结合命令模式让代码更清晰、可测试、易扩展
• 如何应对嵌套事务和复杂业务场景

这不是一个炫技的技术点，而是每一个认真对待数据一致性的工程师都应该掌握的基础能力。无论你是做电商平台、金融系统还是社交网络，只要涉及多表写入，Unit of Work 都是你不可或缺的工具。

记住一句话：

“没有事务的数据库操作，就像没有安全带的汽车——看起来能跑，但随时可能翻车。”

希望今天的分享对你有所启发。如果你正在搭建一个生产级别的 Node.js 应用，请务必把 Unit of Work 加入你的技术栈！