如何在Node.js Worker Threads中共享MongoDB连接池

核心问题在于：Worker Threads是独立的V8执行上下文，无法共享主线程的MongoDB连接；如果每个Worker单独创建连接，会导致连接数随Worker和Pod数量剧增（20个Worker × 50个Pod = 1000+连接），远超MongoDB的默认连接上限。

下面是几种可行的解决方案：

方案一：主线程统一处理数据库操作，Worker仅负责业务计算 #

让所有Worker将数据库查询/写入请求发送给主线程，由主线程统一执行数据库操作后返回结果。这样每个Pod（Node进程）仅需维护1个MongoDB连接，彻底解决连接数爆炸问题。

修改代码示例： #

worker.js（Worker端）

const { parentPort } = require("worker_threads");
const { v4: uuidv4 } = require("uuid");
const threadId = uuidv4();

async function runTask() {
  try {
    while (true) {
      // 向主线程发送查询请求
      parentPort.postMessage({ type: "query", query: { status: "pending" } });
      
      // 等待主线程返回结果
      parentPort.once("message", (result) => {
        console.log(`Thread ID: ${threadId}, Time: ${new Date().toISOString()}, Query Result: ${JSON.stringify(result)}`);
      });

      await new Promise(resolve => setTimeout(resolve, 5000));
    }
  } catch (error) {
    parentPort.postMessage(`Thread ID: ${threadId}, Error: ${error.message}`);
  }
}

runTask();

app.js（主线程）

const { Worker } = require("worker_threads");
const db = require("./db");
const { Migration } = require("./openSearchMigrationModel");

function spawnWorkers(numberOfWorkers) {
  for (let i = 0; i < numberOfWorkers; i++) {
    const worker = new Worker("./worker.js");
    
    worker.on("message", async (msg) => {
      if (msg.type === "query") {
        // 主线程执行数据库查询
        const result = await Migration.find(msg.query);
        // 将结果返回给Worker
        worker.postMessage(result);
      }
    });
    
    worker.on("error", (err) => console.error(`Worker error: ${err.message}`));
  }
}

spawnWorkers(4);

方案二：为每个Pod限制MongoDB连接池大小 #

如果业务场景需要Worker直接操作数据库，可以调整Mongoose的连接池配置，限制每个Pod的总连接数，避免连接数超限。

修改db.js的连接配置： #

const mongoose = require("mongoose");
const mongoURI = "mongodb+srv://xxxxxxxx.mongodb.net/opensearch_migration";

// 配置连接池，限制每个Pod的连接数
mongoose.connect(mongoURI, {
  useNewUrlParser: true,
  useUnifiedTopology: true,
  maxPoolSize: 10, // 每个连接池的最大连接数，根据Pod数量调整
  minPoolSize: 2
});

console.log("Connected to MongoDB");
const db = mongoose.connection;
db.on("error", console.error.bind(console, "MongoDB connection error:"));

module.exports = db;

之后在每个Worker中引入db.js，所有Worker会共用当前Pod的连接池（Worker在同一个进程内的连接池获取连接，不会额外创建新的连接池）。这种方式的连接数上限为 Pod数量 × maxPoolSize，需根据MongoDB的总连接上限合理设置。

方案三：使用任务队列拆分工作 #

如果数据处理任务可以拆分，可引入Redis等任务队列，将待处理的任务放入队列，每个Worker从队列中获取任务处理，主线程负责将任务分发到队列，同时控制并发数。这种方式不仅能减少数据库连接数，还能更好地控制整体任务的执行节奏。

关键思路： #

1. 主线程批量查询pending状态的数据，将任务拆分后推入Redis队列；
2. 每个Worker从Redis队列中拉取任务，处理完成后更新数据状态；
3. 所有Worker共用主线程的数据库连接（或每个Worker复用Pod的连接池）。

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