Typeorm结合EntityManager事务使用LOCK TABLE的技术问题咨询

问题背景 #

使用TypeORM时遇到竞态条件：两个系统几乎同时执行包含「查询+条件插入」的事务，目前通过LOCK TABLE语句手动加表级锁解决，现咨询两个问题：

1. 表级锁是否是多系统并发访问同表事务的正确方案？有没有让PostgreSQL自动排队处理事务的替代方式？
2. 有没有比手动执行em.query('LOCK TABLE ...')更便捷的加锁方式？期望有类似em.lockTable(Entity)的方法。

问题1：锁方案的合理性与替代方案 #

表级锁的定位 #

表级锁是可行但粗粒度的方案：它能确保事务串行化执行，避免竞态，但会阻塞整张表的所有写操作（甚至部分读操作，取决于锁模式），严重影响并发性能，仅适合全表必须串行化的极端场景。

更优替代方案 #

1. UPSERT（推荐）

直接利用PostgreSQL的INSERT ... ON CONFLICT语法，在SQL层面原子性完成「不存在则插入，存在则忽略/更新」，从根源避免竞态，无需额外锁或事务内查询判断。
示例代码：

await em.insert(TestCaseProcessorEntity, testCaseProcessorEntity)
  .onConflict(['id']) // 指定冲突字段（需有唯一约束）
  .doNothing(); // 存在则忽略，也可写doUpdate执行更新逻辑

2. 行级锁（SELECT ... FOR UPDATE）

如果必须保留查询后执行额外逻辑的流程，可在查询时锁定目标行，仅阻塞同一行的并发操作，不影响其他行：

const found = await em.findOne(TestCaseProcessorEntity, {
  where: {id: testCaseProcessorEntity.id},
  select: {id: true, name: true, parameters: {id: true}},
  relations: {parameters: true},
  lock: { mode: "pessimistic_write" } // TypeORM内置的悲观写锁，对应SELECT ... FOR UPDATE
});

3. 串行化隔离级别

将事务隔离级别设为SERIALIZABLE，PostgreSQL会自动确保事务按串行顺序执行，检测到竞态时回滚冲突事务。但需要处理事务重试（不符合你「避免多次重试」的需求，仅作参考）。

问题2：便捷的加锁方式 #

自定义lockTable方法 #

可以通过封装逻辑或扩展TypeORM的EntityManager，实现类似em.lockTable(Entity)的便捷调用：

方式1：工具函数封装

import { EntityManager, ObjectType } from "typeorm";

export async function lockTable<T>(em: EntityManager, entity: ObjectType<T>) {
  const tableName = em.getRepository(entity).metadata.tableName;
  await em.query(`LOCK TABLE "${tableName}"`); // 加引号避免表名含特殊字符
}

使用时直接调用：

await lockTable(em, TestCaseProcessorEntity);

方式2：扩展EntityManager类

如果需要全局可用，可通过TypeORM的类型扩展机制，给EntityManager添加lockTable方法（注意TypeORM版本兼容性）：

import { EntityManager, ObjectType } from "typeorm";

declare module "typeorm" {
  interface EntityManager {
    lockTable<T>(entity: ObjectType<T>): Promise<void>;
  }
}

EntityManager.prototype.lockTable = async function <T>(entity: ObjectType<T>) {
  const tableName = this.getRepository(entity).metadata.tableName;
  await this.query(`LOCK TABLE "${tableName}"`);
};

之后即可直接使用await em.lockTable(TestCaseProcessorEntity)。

优化后的示例代码（UPSERT版本） #

import {EntityManager} from "typeorm";

class TestCaseRegister {
  entityManager: EntityManager;
  logger: Logger;

  async registerProcessor(testCaseProcessorEntity: TestCaseProcessorEntity) {
    try {
      await this.entityManager
        .insert(TestCaseProcessorEntity, testCaseProcessorEntity)
        .onConflict(['id'])
        .doNothing();
      
      this.logger.log('测试用例处理器注册完成（存在则忽略）');
      
      // 如果需要对已存在的实体执行额外逻辑，可后续查询（无需在事务内）
      const found = await this.entityManager.findOne(TestCaseProcessorEntity, {
        where: {id: testCaseProcessorEntity.id},
        select: {id: true, name: true, parameters: {id: true}},
        relations: {parameters: true}
      });
      
      if (found) {
        // 执行额外逻辑
      }
    } catch (err) {
      this.logger.error('注册失败', err);
      throw err;
    }
  }
}

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