Node.js中使用Worker Thread实现SQL Server连接与并发数据查询

核心实现思路 #

• Worker Thread 与主线程内存隔离，数据库连接实例无法直接跨线程传递，必须在每个 Worker 内部独立创建连接或连接池
• 优先使用 Worker 池复用线程，避免单次查询创建销毁线程的额外开销
• Worker 池容量需要和 SQL Server 的最大连接数配置匹配，避免打爆数据库连接

前置依赖 #

使用官方 mssql 驱动连接 SQL Server，用 workerpool 简化 Worker 池调度，执行安装：
npm install mssql workerpool express

1. 编写 Worker 线程执行文件（sql.worker.js） #

Worker 内部独立维护数据库连接池，处理实际查询逻辑：

const sql = require('mssql')

// 数据库配置可从主线程通过初始化参数传入，避免硬编码
const DB_CONFIG = {
  server: 'YOUR_DB_HOST',
  database: 'YOUR_DB_NAME',
  user: 'YOUR_DB_USER',
  password: 'YOUR_DB_PWD',
  options: {
    encrypt: true,
    trustServerCertificate: true
  },
  pool: {
    max: 2,
    min: 0,
    idleTimeoutMillis: 30000
  }
}

let globalPool = null

// 初始化连接池，单Worker内复用
async function getPool() {
  if (!globalPool) {
    globalPool = await sql.connect(DB_CONFIG)
  }
  return globalPool
}

// 暴露给主线程的查询方法
async function execQuery(queryStr, params = []) {
  const pool = await getPool()
  const request = pool.request()
  // 绑定参数防止SQL注入
  params.forEach(p => request.input(p.name, p.type, p.value))
  const res = await request.query(queryStr)
  return res.recordset
}

module.exports = { execQuery }

2. 主线程业务调度代码 #

主线程维护 Worker 池，分发查询任务，实现并发执行：

const workerpool = require('workerpool')
const express = require('express')
const sql = require('mssql')
const app = express()

// 初始化Worker池，maxWorkers根据CPU核心数、数据库最大连接数调整
// 总数据库连接上限 = maxWorkers * 单Worker内连接池max，需小于SQL Server配置的最大连接数
const workerPool = workerpool.pool('./sql.worker.js', {
  minWorkers: 2,
  maxWorkers: 8
})

// 业务API示例
app.get('/api/biz-data', async (req, res) => {
  try {
    // 并发执行多个独立查询
    const [userData, orderData, statsData] = await Promise.all([
      workerPool.exec('execQuery', [
        'SELECT * FROM users WHERE id = @userId',
        [{ name: 'userId', type: sql.Int, value: req.query.userId }]
      ]),
      workerPool.exec('execQuery', [
        'SELECT * FROM orders WHERE user_id = @userId AND status = @status',
        [{ name: 'userId', type: sql.Int, value: req.query.userId }, { name: 'status', type: sql.VarChar(10), value: 'paid' }]
      ]),
      workerPool.exec('execQuery', [
        'SELECT SUM(amount) as total_consume FROM orders WHERE user_id = @userId',
        [{ name: 'userId', type: sql.Int, value: req.query.userId }]
      ])
    ])
    res.json({ code: 200, data: { userData, orderData, statsData } })
  } catch (err) {
    res.status(500).json({ code: 500, msg: '查询失败', error: err.message })
  }
})

// 进程退出时销毁资源
process.on('SIGINT', async () => {
  await workerPool.terminate()
  process.exit(0)
})

app.listen(3000, () => console.log('API服务启动，端口3000'))

关键注意事项 #

• 本身是IO密集型的简单查询不需要用Worker Thread，mssql 自带的异步连接池已经足够，Worker Thread 更适合查询后需要大量CPU计算、多查询并发调度的场景
• 不要跨Worker处理事务，事务绑定单个数据库连接，无法跨线程共享
• 大结果集查询不适合用Worker传递，序列化/反序列化的开销会抵消并发收益，这类场景直接在主线程用流式查询处理即可
• 若不想引入第三方workerpool依赖，用原生worker_threads模块实现逻辑完全一致，仅需自己实现任务队列和Worker复用逻辑即可

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