Node.js for循环执行MySQL查询不等待响应导致重复插入如何解决？

问题根本原因 #

• 常规for循环为同步执行逻辑，不会等待异步的check_by_column回调返回结果就会执行下一轮循环，两次查询请求几乎同时发往MySQL时，目标数据还未插入，两次查询都判定数据不存在，因此触发两次插入操作。
• 即使把循环调整为串行执行，高并发场景下依然存在竞态条件：多次请求的查询操作都在插入操作前完成，仍然会出现重复插入问题。

修复方案 #

方案1：调整异步循环逻辑，串行执行任务 #

先将回调式的数据库方法改造为Promise版本，通过async/await控制异步流程：

db_connection.js改造

// 改造check_by_column为Promise版本
check_by_column(tablename, selectedKeys, values, selectedValues) {
  return new Promise((resolve, reject) => {
    const show_key = [];
    for (const [key, value] of Object.entries(values)) {
      show_key.push(`${key} = '${value}'`);
    }
    // 原代码拼接条件用空格是错误逻辑，多条件需要用AND连接
    const column = show_key.join(" AND ");
    const sql = `SELECT ${selectedKeys} FROM ${tablename} WHERE ${column}`;
    console.log(sql);
    con.query(sql, function (err, result) {
      if (err) {
        console.log("database Error");
        return reject(err);
      }
      resolve(result && result.length === 0 ? 5 : 3);
    })
  })
}

// 改造insert_value为Promise版本
insert_value(tablename, values) {
  return new Promise((resolve, reject) => {
    const show_key = [];
    const show_Values = [];
    for (const [key, value] of Object.entries(values)) {
      show_key.push(key);
      show_Values.push(value);
    }
    const nKey = show_key.join(", ");
    const nValue = show_Values.map(v => `'${v}'`).join(", ");
    const sql = `INSERT INTO ${tablename} (${nKey}) VALUES (${nValue})`;
    con.query(sql, function (err, result) {
      if (err) {
        console.log("database Error", err);
        return reject(err);
      }
      resolve(result);
    });
  })
}

// 按照相同逻辑把update_by_column也改造为Promise版本

index.js改造

var dat = {"IMEI": 80, "rule": 2, "id": 1, "operation": 4, "address": 48, "qty": 6, "delay": 10, "data" : "3231,3039,3339", "error" : 0, "Timestamp":"2021-10-12 11:41:22"};
const topicData = JSON.stringify(dat);
var query = JSON.parse(topicData);
const currentDateTime = moment().format("YYYY-MM-DD HH:mm:ss");

const groups = [1, 2];
// 用async自执行函数包裹逻辑，通过await串行执行异步任务
(async () => {
  for (let i = 0; i < groups.length; i++) {
    console.log(i);
    const response = await db.check_by_column('ml', 'number', { "number": query.data }, {"topic":"80", "IMEI":query.IMEI, "id":query.id, "number": query.data, "start_date":currentDateTime, "end_date": currentDateTime});
    if (response === 5) {
      await db.insert_value('ml', {"topic":"80", "IMEI":query.IMEI, "id":query.id, "number": query.data, "start_date":currentDateTime, "end_date": currentDateTime});
      console.log("insert");
    } else {
      await db.update_by_column('ml', {"topic":"80", "IMEI":query.IMEI, "id":query.id, "number": query.data, "start_date":currentDateTime, "end_date": currentDateTime}, {"number":query.data});
      console.log("update");
    }
  }
})()

方案2：使用MySQL原生UPSERT（推荐，从根本解决竞态问题） #

不需要先查询再判断插入/更新，直接用MySQL的INSERT ... ON DUPLICATE KEY UPDATE语法，前提是先给ml表的number字段加唯一索引：

第一步执行SQL加唯一索引

ALTER TABLE ml ADD UNIQUE INDEX idx_number (number);

第二步新增UPSERT方法到db_connection.js

upsert_ml(data) {
  return new Promise((resolve, reject) => {
    // 用参数化查询避免SQL注入风险
    const sql = `INSERT INTO ml (topic, IMEI, id, number, start_date, end_date)
    VALUES (?, ?, ?, ?, ?, ?)
    ON DUPLICATE KEY UPDATE
    topic = VALUES(topic), IMEI = VALUES(IMEI), id = VALUES(id), start_date = VALUES(start_date), end_date = VALUES(end_date)`;
    const params = [data.topic, data.IMEI, data.id, data.number, data.start_date, data.end_date];
    con.query(sql, params, (err, result) => {
      if (err) return reject(err);
      resolve(result.affectedRows === 1 ? 'insert' : 'update');
    })
  })
}

第三步改造index.js调用逻辑

(async () => {
  const upsertData = {"topic":"80", "IMEI":query.IMEI, "id":query.id, "number": query.data, "start_date":currentDateTime, "end_date": currentDateTime};
  for (let i = 0; i < groups.length; i++) {
    const opType = await db.upsert_ml(upsertData);
    console.log(opType);
  }
})()

注意事项 #

• 原代码直接拼接SQL语句的写法存在严重SQL注入风险，务必改为参数化查询实现。
• 原check_by_column方法中WHERE条件拼接用join(" ")是错误逻辑，多查询条件需要用AND连接。

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