MongoDB插入超时问题：MySQL转MongoDB脚本执行Community.create报错

问题背景 #

用户定义了MongoDB的Community Schema，编写了从MySQL向MongoDB迁移数据的脚本，脚本可正常连接MongoDB并查询MySQL数据，但执行Community.create时出现以下错误：

Failed to update MongoDB: MongooseError: Operation communities.insertOne() buffering timed out after 10000ms

相关代码 #

community.model.js中的Schema定义

const mongoose = require("mongoose");

const CommunitySchema = new mongoose.Schema(
  {
    admin: {
      type: mongoose.Types.ObjectId,
      ref: "user",
    },
    imdb: {
      type: String,
      unique: true,
      maxLength: [25, "Max length 25"]
    },
    poster: {
      type: String,
      maxLength: [250, "Max length 250"]
    },
    name: {
      type: String,
      required: true,
      unique: true,
      //  minlength: [4, "Must be at least 4 characters long"],
      // maxlength: [30, "Must be no more than 30 characters long"],
    },
    bio: {
      type: String,
      default: "",
      // maxLength: [250, "Must be at most 250 characters long"],
    },
    posts: [
      {
        type: mongoose.Schema.Types.ObjectId,
        ref: "post",
      },
    ],
    subscribedBy: [
      {
        type: mongoose.Schema.Types.ObjectId,
        ref: "user",
      },
    ],
    subscriberCount: {
      type: Number,
      default: 1,
    },
  },
  {
    timestamps: true,
  }
);

module.exports = mongoose.model("community", CommunitySchema);

数据迁移脚本

const mongoose = require("mongoose");
const Community = require("../model/community.model");
const mysql = require("mysql2");
const config = require("./config/config");

const mysqlConnection = mysql.createConnection({
  host: config.database.host,
  user: config.database.user,
  password: config.database.password,
  database: config.database.databaseName,
});
(async() => {
  await db()

 })()

const schemaFields = config.mysqlQueryColumns.reduce((acc, field) => {
  acc[field.name] = field.type;
  return acc;
}, {});


const updateMongo = async () => {

  try {
    
    if (mongoose.connection.readyState !== 1) {

      console.log("MongoDB connection is not open. Trying to connect...");
      await mongoose.connect(config.database.mongodb);
 
   
      console.log("Connected to MongoDB");
    }
   
    const columns = config.mysqlQueryColumns.map((field) => field.name).join(", ");

    mysqlConnection.query(`SELECT ${columns} FROM ${config.table_name} LIMIT 10`,(err, results) => {

        if (err) {
          console.error(`MySQL query error: ${err}`);
          return;
        }

        results.forEach(async (row) => {
          try {
           
            let updateData = {};
            config.mysqlQueryColumns.forEach((field) => {
              updateData[field.name] = row[field.name];

            });
          console.log(updateData);
          
           await Community.create({

                name : updateData.titol,
                bio: updateData.event,
                subscribedBy : ["661d5aef399b7e21da05b2a1"],
                admin: "661d5aef399b7e21da05b2a1"
          //     admin: userId,
          //      subscribedBy: [userId],
              
           });
       
          } catch (err) {
            console.error(`Failed to update MongoDB: ${err}`);
          }
        });
        
        console.log("Data updated to MongoDB");
      });
  } catch (err) {
    console.error("Failed to update MongoDB:", err);
  }
};

async function db() {
await  mongoose.connect(config.database.mongodb).then(() => {
      console.log("Connected to MongoDB");
     
      updateMongo();
    })
    .catch((err) => {
      console.error("MongoDB connection error:", err);
    });
}

setInterval(() => {
  console.log("Updating MongoDB...");
  if (mongoose.connection.readyState !== 1) {
   db();

  } else {
    updateMongo();
  }
}, config.fetch_time || 10 * 1000 * 60);

MySQL记录示例

{
  imdb: 'tt0082533',
  titol: 'Aullidos (1981)',
  poster: 'https://www.themoviedb.org/t/p/original/dEFMku7o825YhyUEHSlkvLDz1Dt.jpg',
  event: 'After a bizarre and near fatal encounter with a serial killer, a newswoman is sent to a rehabilitation center whose inhabitants may not be what they seem.'
}

问题原因 #

1. 异步回调与连接状态冲突：mysqlConnection.query是回调式API，回调内执行Community.create时，Mongoose连接可能未完全就绪，导致操作被缓冲后超时。
2. 重复连接逻辑混乱：脚本中多处触发MongoDB连接（自执行函数、updateMongo、定时器），连接状态重叠，可能出现未建立连接就执行插入的情况。
3. forEach异步操作未等待：results.forEach(async (row) => {})中的异步操作不会被等待，大量并发插入超出Mongoose缓冲限制。

修复方案 #

1. 统一MongoDB连接管理 #

重构连接函数，避免重复连接，确保连接就绪后再执行操作：

async function db() {
  if (mongoose.connection.readyState === 1) {
    return Promise.resolve(); // 已连接直接返回
  }
  try {
    await mongoose.connect(config.database.mongodb);
    console.log("Connected to MongoDB");
  } catch (err) {
    console.error("MongoDB connection error:", err);
    throw err;
  }
}

2. 封装MySQL查询为Promise #

将回调式查询改为Promise，配合async/await保证流程顺序：

const queryMySQL = (sql) => {
  return new Promise((resolve, reject) => {
    mysqlConnection.query(sql, (err, results) => {
      if (err) reject(err);
      else resolve(results);
    });
  });
};

3. 修复updateMongo的异步逻辑 #

确保连接就绪后查询MySQL，用for...of替代forEach等待所有插入完成，同时补充Schema要求的字段：

const updateMongo = async () => {
  try {
    await db();

    const columns = config.mysqlQueryColumns.map((field) => field.name).join(", ");
    const results = await queryMySQL(`SELECT ${columns} FROM ${config.table_name} LIMIT 10`);

    for (const row of results) {
      try {
        const updateData = config.mysqlQueryColumns.reduce((acc, field) => {
          acc[field.name] = row[field.name];
          return acc;
        }, {});
        
        console.log(updateData);
        
        await Community.create({
          name: updateData.titol,
          imdb: updateData.imdb, // 补充unique字段imdb，避免重复插入报错
          poster: updateData.poster, // 补充Schema定义的poster字段
          bio: updateData.event,
          subscribedBy: ["661d5aef399b7e21da05b2a1"],
          admin: "661d5aef399b7e21da05b2a1"
        });
      } catch (err) {
        console.error(`Failed to insert row: ${err}`);
      }
    }

    console.log("Data updated to MongoDB");
  } catch (err) {
    console.error("Failed to update MongoDB:", err);
  }
};

4. 简化初始化与定时器逻辑 #

去掉冗余的自执行函数，统一调用updateMongo：

// 初始化执行一次
(async () => {
  await updateMongo();
})();

// 定时器逻辑
setInterval(async () => {
  console.log("Updating MongoDB...");
  await updateMongo();
}, config.fetch_time || 10 * 1000 * 60);

额外注意事项 #

• 检查subscribedBy和admin传入的ObjectId是否在user集合中存在，避免关联错误；
• 插入前可通过imdb或name查询是否已存在记录，避免触发unique约束错误；
• 若数据量较大，可添加批量插入逻辑，减少数据库请求次数。

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