Sequelize批量更新百万级数据丢失行及字段为null问题求助

原实现存在几个核心问题，直接导致了更新异常：

• 串行更新无错误处理：循环内逐个执行await tranz.update，一旦某条更新抛出错误（比如数据库连接中断、字段校验失败），后续代码直接终止，该批次剩余记录未更新且无日志提示。
• 递归调用未做异步控制：runBatch递归调用时未等待当前批次更新完成，可能导致多批次并发执行，耗尽数据库连接池，引发更新超时或失败。
• 变量未做null校验：计算后的number变量可能在边界场景下意外变为null（比如空值运算、异常计算逻辑），直接赋值给quantity导致字段被置空。
• 单条更新效率极低：60万条记录单条更新会产生60万次数据库请求，极大占用资源，容易触发数据库连接限制或超时机制。

1. 基础修复：添加错误捕获与异步控制 #

先给原代码补全错误处理，确保单条更新失败不中断批次，同时控制递归执行顺序：

runBatch(offset){
  let limit = 100000;
  db.sequelize.models.transactions.schema("queenDB").findAll({
      limit: limit,
      offset: offset,
      order: [['txndate', 'asc']]
  }).then(async(tranzObj)=> {
    for (let tranz of tranzObj) {
      try {
        /* some calculations */
        // 强制校验number非空，避免赋值null
        if (number == null) {
          console.error(`无效quantity值，交易ID：${tranz.id}，值：${number}`);
          continue;
        }
        await tranz.update({ quantity: number, dateupdated: new Date() });
      } catch (err) {
        console.error(`更新交易失败，ID：${tranz.id}，错误：`, err.message);
      }
    }
    
    // 等待当前批次完成后再递归，避免并发过载
    if (tranzObj.length === limit) {
      await runBatch(offset + limit);
    } else {
      console.log("批量更新完成");
    }
  }).catch(err => {
    console.error(`获取批次数据失败，偏移量：${offset}，错误：`, err.message);
  });
}

2. 最优方案：改用批量更新提升效率 #

单条更新效率过低，建议先批量查询记录、计算更新值，再用bulkUpdate一次性执行，大幅减少数据库请求次数：

async function runBatch(offset) {
  const limit = 10000; // 缩小批次大小，避免内存占用过高
  try {
    // 1. 查询当前批次记录，只获取计算和更新需要的字段
    const tranzObj = await db.sequelize.models.transactions.schema("queenDB").findAll({
      limit: limit,
      offset: offset,
      order: [['txndate', 'asc']],
      attributes: ['id', /* 计算逻辑需要的其他字段 */]
    });

    if (tranzObj.length === 0) {
      console.log("批量更新完成");
      return;
    }

    // 2. 批量计算需要更新的数据
    const updateData = [];
    for (const tranz of tranzObj) {
      /* some calculations */
      if (number == null) {
        console.error(`无效quantity值，交易ID：${tranz.id}，值：${number}`);
        continue;
      }
      updateData.push({
        id: tranz.id,
        quantity: number,
        dateupdated: new Date()
      });
    }

    // 3. 执行批量更新
    if (updateData.length > 0) {
      await db.sequelize.models.transactions.schema("queenDB").bulkUpdate(
        updateData,
        { id: db.Sequelize.where(db.Sequelize.col('id'), 'IN', updateData.map(item => item.id)) },
        { individualHooks: false } // 不需要模型钩子可关闭，提升速度
      );
    }

    // 4. 递归执行下一批次
    if (tranzObj.length === limit) {
      await runBatch(offset + limit);
    } else {
      console.log("批量更新完成");
    }
  } catch (err) {
    console.error(`批次更新失败，偏移量：${offset}，错误：`, err.message);
    // 可选：添加当前批次重试逻辑
    // await new Promise(resolve => setTimeout(resolve, 5000));
    // await runBatch(offset);
  }
}

3. 避免递归栈溢出：改用循环迭代 #

如果批次数量过多（60万/1万=60批），递归调用可能导致栈溢出，建议改用循环迭代：

async function runAllBatches() {
  const limit = 10000;
  let offset = 0;
  let hasMore = true;

  while (hasMore) {
    try {
      const tranzObj = await db.sequelize.models.transactions.schema("queenDB").findAll({
        limit: limit,
        offset: offset,
        order: [['txndate', 'asc']],
        attributes: ['id', /* 计算字段 */]
      });

      if (tranzObj.length === 0) {
        hasMore = false;
        break;
      }

      // 计算并批量更新
      const updateData = [];
      for (const tranz of tranzObj) {
        /* some calculations */
        if (number == null) {
          console.error(`无效quantity值，交易ID：${tranz.id}，值：${number}`);
          continue;
        }
        updateData.push({
          id: tranz.id,
          quantity: number,
          dateupdated: new Date()
        });
      }

      if (updateData.length > 0) {
        await db.sequelize.models.transactions.schema("queenDB").bulkUpdate(
          updateData,
          { id: { [db.Sequelize.Op.in]: updateData.map(item => item.id) } }
        );
      }

      offset += limit;
      console.log(`已完成批次，偏移量：${offset - limit}`);
    } catch (err) {
      console.error(`批次更新失败，偏移量：${offset}，错误：`, err.message);
    }
  }
  console.log("所有批次更新完成");
}

4. 数据库层面防护 #

• 给quantity字段添加NOT NULL约束，即使代码出错，数据库也会拒绝null值的更新，避免数据被意外置空。
• 调整Sequelize连接池配置，设置pool.max为合适值，确保有足够连接处理批量操作。

5. 可选：事务控制保证原子性 #

如果需要保证单个批次的更新原子性（要么全成功，要么全回滚），可以用事务包裹批量更新：

const transaction = await db.sequelize.transaction();
try {
  await db.sequelize.models.transactions.schema("queenDB").bulkUpdate(..., { transaction });
  await transaction.commit();
} catch (err) {
  await transaction.rollback();
  throw err;
}

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