nestjs 事务处理

'# NestJS 事务处理

一、背景与问题

在分布式系统中，事务处理是保障数据一致性的核心机制。NestJS 作为流行的 Node.js 框架，其事务处理能力依赖于底层数据库驱动的实现（如 TypeORM、MongoDB 等）。然而，开发者常因对事务边界、传播机制和异常处理的误解，导致数据不一致或性能问题。

典型场景包括：

转账业务（扣款-入账）
订单创建（库存扣减-订单记录）
多表关联的业务操作

传统做法中，开发者容易犯以下错误：

忘记在服务层显式声明事务边界
在事务中调用非事务性方法导致部分操作未参与事务
忽略事务的隔离级别对并发性能的影响

二、基本原理

NestJS 的事务处理本质是通过装饰器和拦截器机制，将事务上下文传递到数据库驱动层。其核心机制包含三个层次：

装饰器声明：通过 @Transaction() 装饰器标记事务边界
拦截器处理：在方法调用时创建事务上下文
驱动层执行：将事务上下文传递给数据库驱动（如 TypeORM、MongoDB）

事务的 ACID 特性在 NestJS 中表现为：

原子性：通过 BEGIN 和 COMMIT/ROLLBACK 确保操作要么全成功要么全回滚
一致性：通过事务边界确保数据状态的一致性
隔离性：通过设置不同的隔离级别（READ COMMITTED/REPEATABLE READ）控制并发访问
持久性：通过事务提交将变更永久保存到数据库

三、环境准备

npm install @nestjs/common @nestjs/core @nestjs/platform-express
npm install typeorm mysql2
npm install --save-dev @types/mysql2

四、核心实现

1. 基础事务处理

// transaction.service.ts
import { Injectable, Transaction } from '@nestjs/common';
import { Repository } from 'typeorm';
import { InjectRepository } from '@nestjs/typeorm';
import { Account } from './entities/account.entity';

@Injectable()
export class TransactionService {
  constructor(
    @InjectRepository(Account)
    private readonly accountRepository: Repository<Account>,
  ) {}

  @Transaction()
  async transfer(fromId: number, toId: number, amount: number) {
    const fromAccount = await this.accountRepository.findOne({
      where: { id: fromId },
    });
    
    const toAccount = await this.accountRepository.findOne({
      where: { id: toId },
    });

    // 事务上下文已自动创建
    await this.accountRepository.update(fromId, {
      balance: fromAccount.balance - amount,
    });

    await this.accountRepository.update(toId, {
      balance: toAccount.balance + amount,
    });
  }
}

关键代码解释：

@Transaction() 装饰器声明事务边界
update() 操作会自动参与到当前事务中
事务提交/回滚由框架自动管理

2. 事务传播控制

// transaction.service.ts
import { Injectable, Transaction, InjectRepository } from '@nestjs/common';
import { Repository } from 'typeorm';
import { Account } from './entities/account.entity';

@Injectable()
export class TransactionService {
  constructor(
    @InjectRepository(Account)
    private readonly accountRepository: Repository<Account>,
  ) {}

  @Transaction()
  async transfer(fromId: number, toId: number, amount: number) {
    const fromAccount = await this.accountRepository.findOne({
      where: { id: fromId },
    });
    
    const toAccount = await this.accountRepository.findOne({
      where: { id: toId },
    });

    // 事务传播：当前事务上下文传递给子方法
    await this.withdraw(fromId, amount);
    
    await this.deposit(toId, amount);
  }

  async withdraw(id: number, amount: number) {
    // 由于未使用 @Transaction 装饰器，此方法不在事务中
    // 需要显式获取事务上下文
    const transaction = this.getTransactionContext();
    
    if (!transaction) {
      throw new Error('Transaction context not available');
    }

    await transaction.manager.update(id, {
      balance: this.accountRepository.findOne(id).balance - amount,
    });
  }

  async deposit(id: number, amount: number) {
    const transaction = this.getTransactionContext();
    
    if (!transaction) {
      throw new Error('Transaction context not available');
    }

    await transaction.manager.update(id, {
      balance: this.accountRepository.findOne(id).balance + amount,
    });
  }

  getTransactionContext(): any {
    // 获取当前事务上下文（具体实现依赖于数据库驱动）
    return this.accountRepository.manager.getTransactionContext();
  }
}

关键代码解释：

事务传播机制需要显式获取事务上下文
子方法需要通过 transaction.manager 执行操作
未使用事务装饰器的方法默认不参与事务

3. 异常处理与回滚

// transaction.service.ts
import { Injectable, Transaction, HttpException, HttpStatus } from '@nestjs/common';
import { Repository } from 'typeorm';
import { Account } from './entities/account.entity';

@Injectable()
export class TransactionService {
  constructor(
    @InjectRepository(Account)
    private readonly accountRepository: Repository<Account>,
  ) {}

  @Transaction()
  async transfer(fromId: number, toId: number, amount: number) {
    try {
      const fromAccount = await this.accountRepository.findOne({
        where: { id: fromId },
      });
      
      const toAccount = await this.accountRepository.findOne({
        where: { id: toId },
      });

      await this.accountRepository.update(fromId, {
        balance: fromAccount.balance - amount,
      });

      // 模拟异常
      if (amount > 1000) {
        throw new HttpException('Transfer amount too large', HttpStatus.BAD_REQUEST);
      }

      await this.accountRepository.update(toId, {
        balance: toAccount.balance + amount,
      });

      return { success: true };
    } catch (error) {
      // 事务自动回滚
      throw error;
    }
  }
}

关键代码解释：

异常会触发事务回滚
需要显式捕获异常并重新抛出
框架会自动处理事务提交/回滚

五、完整案例

1. 订单创建与库存扣减

// order.controller.ts
import { Controller, Post, Body, HttpCode, HttpStatus } from '@nestjs/common';
import { OrderService } from './order.service';
import { CreateOrderDto } from './dto/create-order.dto';

@Controller('orders')
export class OrderController {
  constructor(private readonly orderService: OrderService) {}

  @Post()
  @HttpCode(HttpStatus.CREATED)
  async createOrder(@Body() createOrderDto: CreateOrderDto) {
    return await this.orderService.createOrder(createOrderDto);
  }
}

// order.service.ts
import { Injectable, Transaction } from '@nestjs/common';
import { Repository } from 'typeorm';
import { InjectRepository } from '@nestjs/typeorm';
import { Order } from './entities/order.entity';
import { Product } from './entities/product.entity';

@Injectable()
export class OrderService {
  constructor(
    @InjectRepository(Order)
    private readonly orderRepository: Repository<Order>,
    @InjectRepository(Product)
    private readonly productRepository: Repository<Product>,
  ) {}

  @Transaction()
  async createOrder(createOrderDto: CreateOrderDto) {
    const { productId, quantity } = createOrderDto;

    const product = await this.productRepository.findOne({
      where: { id: productId },
    });

    if (!product || product.stock < quantity) {
      throw new HttpException('Insufficient stock', HttpStatus.BAD_REQUEST);
    }

    // 扣减库存
    await this.productRepository.update(productId, {
      stock: product.stock - quantity,
    });

    // 创建订单
    const order = this.orderRepository.create({
      productId,
      quantity,
      total: product.price * quantity,
    });

    await this.orderRepository.save(order);

    return order;
  }
}

完整案例说明：

事务覆盖了库存扣减和订单创建两个操作
如果库存不足或保存订单失败，事务会自动回滚
使用 save() 而不是 update() 保证完整性

六、源码解析

以 TypeORM 的事务处理为例，其底层通过 EntityManager 实现事务控制：

// TypeORM 事务实现（简化版）
async function beginTransaction(): Promise<void> {
  await this.connection.driver.beginTransaction();
}

async function commitTransaction(): Promise<void> {
  await this.connection.driver.commitTransaction();
}

async function rollbackTransaction(): Promise<void> {
  await this.connection.driver.rollbackTransaction();
}

关键点：

事务边界由 @Transaction() 装饰器创建
框架通过 EntityManager 传递事务上下文
所有数据库操作都通过 EntityManager 执行

七、进阶使用

1. 多数据库事务支持

// multi-database.service.ts
import { Injectable, Transaction, InjectRepository } from '@nestjs/common';
import { Repository } from 'typeorm';
import { Account } from './entities/account.entity';
import { Product } from './entities/product.entity';

@Injectable()
export class MultiDatabaseService {
  constructor(
    @InjectRepository(Account)
    private readonly accountRepository: Repository<Account>,
    @InjectRepository(Product)
    private readonly productRepository: Repository<Product>,
  ) {}

  @Transaction()
  async transferAndOrder(fromId: number, toId: number, amount: number) {
    const fromAccount = await this.accountRepository.findOne({
      where: { id: fromId },
    });
    
    const toAccount = await this.accountRepository.findOne({
      where: { id: toId },
    });

    await this.accountRepository.update(fromId, {
      balance: fromAccount.balance - amount,
    });

    await this.productRepository.update(1, {
      stock: this.productRepository.findOne(1).stock - 10,
    });
  }
}

2. 事务传播模式

// transaction.service.ts
import { Injectable, Transaction, InjectRepository } from '@nestjs/common';
import { Repository } from 'typeorm';
import { Account } from './entities/account.entity';

@Injectable()
export class TransactionService {
  constructor(
    @InjectRepository(Account)
    private readonly accountRepository: Repository<Account>,
  ) {}

  @Transaction()
  async transfer(fromId: number, toId: number, amount: number) {
    // 传播模式1：REQUIRED（默认）
    await this.withdraw(fromId, amount);
    
    // 传播模式2：NEVER
    await this.deposit(toId, amount);
  }

  async withdraw(id: number, amount: number) {
    const transaction = this.getTransactionContext();
    
    if (!transaction) {
      throw new Error('Transaction context not available');
    }

    await transaction.manager.update(id, {
      balance: this.accountRepository.findOne(id).balance - amount,
    });
  }

  async deposit(id: number, amount: number) {
    const transaction = this.getTransactionContext();
    
    if (!transaction) {
      throw new Error('Transaction context not available');
    }

    await transaction.manager.update(id, {
      balance: this.accountRepository.findOne(id).balance + amount,
    });
  }
}

八、性能与工程实践

1. 性能优化方法

事务粒度控制：避免长事务，按最小业务单元划分事务
批量操作：使用 saveMany() 替代多个 save() 调用
索引优化：确保事务中使用的字段有索引
连接池配置：根据业务量调整数据库连接池大小

2. 异常处理策略

重试机制：对可重试的异常（如网络问题）进行重试
补偿事务：对不可重试的异常执行补偿操作（如发送通知）
日志记录：记录事务日志以便排查问题

3. 安全风险

SQL注入：使用参数化查询避免直接拼接 SQL
事务污染：避免在事务中执行非事务性操作
权限控制：确保事务操作符合权限要求

九、常见问题与踩坑

1. 常见错误

错误	原因	解决方案
事务未提交	忘记在方法上使用 `@Transaction` 装饰器	添加事务装饰器
部分操作未参与事务	在事务中调用非事务性方法	使用 `transaction.manager` 执行操作
事务隔离级别导致死锁	高并发场景下未设置合适的隔离级别	调整隔离级别（如 `READ COMMITTED`）
长事务导致性能问题	事务未及时提交	优化事务粒度，按业务单元划分

2. 典型问题分析

问题：事务在分布式系统中失效

// 错误示例
async function processOrder(order: Order) {
  await saveOrder(order); // 非事务性操作
  await updateInventory(order); // 事务性操作
}

原因：saveOrder 操作未参与事务，导致数据不一致

改进方案：

@Transaction()
async function processOrder(order: Order) {
  await this.orderRepository.save(order); // 事务性操作
  await this.inventoryRepository.update(...);
}

十、最佳实践

事务边界明确：每个事务对应单一业务逻辑
使用事务传播模式：根据需求选择 REQUIRED/NEVER 等模式
异常处理完善：捕获异常并进行适当的回滚
日志记录：记录事务开始/结束时间，便于排查问题
性能监控：监控事务执行时间，避免长事务
安全控制：确保事务操作符合权限要求

十一、总结

NestJS 的事务处理是实现数据一致性的重要机制，其核心在于通过装饰器和拦截器管理事务上下文。开发者需要理解事务的传播机制、异常处理策略和性能影响，才能在实际项目中正确使用。

适用场景：

需要保证数据一致性的核心业务
多表关联的复杂操作
跨服务的分布式事务（需配合分布式事务框架）

不适用场景：

读操作（无需保证一致性）
高并发场景下可能导致锁竞争
轻量级的单次操作

通过合理使用事务处理，可以在保证数据一致性的同时，避免性能瓶颈。在实际开发中，建议结合监控工具和日志分析，持续优化事务处理策略。

nestjs 事务处理

一、背景与问题

二、基本原理

三、环境准备

四、核心实现

1. 基础事务处理

2. 事务传播控制

3. 异常处理与回滚

五、完整案例

1. 订单创建与库存扣减

六、源码解析

七、进阶使用

1. 多数据库事务支持

2. 事务传播模式

八、性能与工程实践

1. 性能优化方法

2. 异常处理策略

3. 安全风险

九、常见问题与踩坑

1. 常见错误

2. 典型问题分析

十、最佳实践

十一、总结

评论已关闭

推荐阅读