TypeORM中处理海量关联数据批量插入的正确方式？

你的问题出在一次性处理超大量关联实体时，TypeORM的save方法会执行拓扑排序（即报错中的SubjectTopoligicalSorter）来处理实体依赖关系，递归深度过大直接导致栈溢出，同时内存占用过高也会拖慢插入速度。以下是两种高效可行的解决方案，能满足Azure S1环境30秒内完成插入的要求：

方案一：使用QueryBuilder原生批量插入（最优选择） #

直接通过QueryBuilder生成原生INSERT语句，跳过TypeORM的实体关系校验、拓扑排序等额外开销，是性能最高的方式。

步骤1：批量插入Sack并获取ID #

先插入所有麻袋，通过returning获取插入后的自增ID（需要数据库支持，比如PostgreSQL、MySQL 8.0+、SQL Server）：

// 提取Sack的基础数据（不包含fruits关联）
const sackData = data.map(sackItem => ({
  // 填充Sack实体的字段，比如name、weight等
}));

// 批量插入Sack并返回插入后的实体（含ID）
const insertedSacks = await getRepository(Sack)
  .createQueryBuilder()
  .insert()
  .into(Sack)
  .values(sackData)
  .returning("*")
  .execute()
  .then(result => result.raw);

步骤2：批量插入关联的Fruit #

根据返回的Sack ID，批量生成Fruit数据并分批次插入：

const fruitData = [];
// 关联每个Fruit对应的Sack ID
insertedSacks.forEach((sack, index) => {
  const originalFruits = data[index];
  const mappedFruits = originalFruits.map(fruit => ({
    ...fruit, // 填充Fruit的字段，比如fruitTypeId、name等
    sackId: sack.id // 关联对应的Sack ID
  }));
  fruitData.push(...mappedFruits);
});

// 分批次插入Fruit，每批10000条（可根据数据库性能调整）
const batchSize = 10000;
for (let i = 0; i < fruitData.length; i += batchSize) {
  const batch = fruitData.slice(i, i + batchSize);
  await getRepository(Fruit)
    .createQueryBuilder()
    .insert()
    .into(Fruit)
    .values(batch)
    .execute();
}

优化：加入事务 #

将所有插入操作包裹在事务中，保证原子性的同时提升插入效率：

await getConnection().transaction(async manager => {
  // 插入Sack
  const insertedSacks = await manager
    .createQueryBuilder()
    .insert()
    .into(Sack)
    .values(sackData)
    .returning("*")
    .execute()
    .then(res => res.raw);

  // 分批次插入Fruit
  const batchSize = 10000;
  for (let i = 0; i < fruitData.length; i += batchSize) {
    const batch = fruitData.slice(i, i + batchSize);
    await manager
      .createQueryBuilder()
      .insert()
      .into(Fruit)
      .values(batch)
      .execute();
  }
});

方案二：分批次使用save方法（兼容旧版TypeORM） #

如果不想改用QueryBuilder，可以将数据拆分成小批次处理，同时利用TypeORM的chunk参数自动分块插入，避免栈溢出：

const batchSize = 500; // 每批处理500个Sack（可调整）
await getConnection().transaction(async manager => {
  for (let i = 0; i < data.length; i += batchSize) {
    const batch = data.slice(i, i + batchSize);
    
    let fruitEntities = [];
    let sackEntities = batch.map(sack => {
      const sackEntity = new Sack();
      // 设置Sack的基础字段
      sackEntity.fruits = sack.map(fruit => CreateFruitEntity());
      fruitEntities = fruitEntities.concat(sackEntity.fruits);
      return sackEntity;
    });
    
    // 分块保存Fruit和Sack，chunk参数指定每块插入的数量
    await manager.getRepository(Fruit).save(fruitEntities, { chunk: 1000 });
    await manager.getRepository(Sack).save(sackEntities, { chunk: 1000 });
  }
});

额外优化建议 #

• 关闭TypeORM的日志输出（如果开启），减少IO开销；
• 调整数据库连接池大小，Azure S1环境建议设置为10-20；
• 确保Fruit和Sack表的外键、索引配置合理，避免插入时的索引维护开销过大。

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