Node.js多用户WebSocket与SQL Server连接方案咨询

核心结论 #

是的，你必须为每个用户创建独立的WebSocket和SQL Server连接实例，因为每个用户拥有专属的资源，单例模式无法满足多用户隔离的需求。下面是基于你现有代码的改造方案：

1. 重构用户配置存储 #

把原来存在config.json里的单用户配置，迁移到MongoDB的User模型中，每个用户存储自己的数据库和WebSocket配置：

// models/user.js
const mongoose = require('mongoose');

const userSchema = new mongoose.Schema({
  // 原有字段
  email: { type: String, required: true },
  password: { type: String, required: true },
  // 添加用户专属配置字段
  dbConfig: {
    user: String,
    password: String,
    server: String,
    database: String
  },
  wsConfig: {
    wssUrl: String,
    tenantId: String,
    tenantSecret: String
  }
});

module.exports = mongoose.model('User', userSchema);

2. 创建连接管理模块 #

新增connectionManager.js统一维护所有用户的连接实例，避免内存泄漏和资源混乱：

// connectionManager.js
const sql = require('mssql');
const WebSocket = require('ws');
const custom = require('./util/custom');

// 存储用户连接：key=userId，value={ dbConn, wsConn }
const userConnections = new Map();

// 初始化单个用户的数据库连接
async function initDbConnection(userId, dbConfig) {
  try {
    const pool = await sql.connect({
      ...dbConfig,
      trustServerCertificate: true
    });
    console.log(`用户 ${userId} 数据库连接成功`);
    return pool;
  } catch (err) {
    console.error(`用户 ${userId} 数据库连接失败:`, err);
    throw err;
  }
}

// 初始化单个用户的WebSocket连接
function initWsConnection(userId, wsConfig, dbPool) {
  const ws = new WebSocket(wsConfig.wssUrl, {
    headers: {
      'tenant-id': wsConfig.tenantId,
      'tenant-secret': wsConfig.tenantSecret
    }
  });

  ws.on('open', () => {
    console.log(`用户 ${userId} WebSocket连接建立`);
    // 定时ping保持连接
    setInterval(() => ws.ping(''), 30000);
  });

  ws.on('message', (data) => {
    const parsedData = JSON.parse(data);
    // 处理消息时传入当前用户的数据库连接
    if (parsedData.action === 'NewMessage') {
      custom.syncMessages(parsedData.messageId, dbPool);
    } else if (parsedData.action === 'SyncHouses') {
      custom.syncHouses(dbPool);
    } else if (parsedData.action === 'SyncMap') {
      custom.syncMap(dbPool);
    }
  });

  ws.on('close', () => {
    console.log(`用户 ${userId} WebSocket连接断开，尝试重连`);
    // 重连逻辑：使用用户配置重新创建连接
    setTimeout(() => {
      const newWs = initWsConnection(userId, wsConfig, dbPool);
      userConnections.set(userId, { ...userConnections.get(userId), wsConn: newWs });
    }, 2000);
  });

  ws.on('error', (err) => {
    console.error(`用户 ${userId} WebSocket错误:`, err);
  });

  return ws;
}

// 初始化单个用户的所有连接
async function initUserConnections(user) {
  const { _id: userId, dbConfig, wsConfig } = user;
  if (!dbConfig || !wsConfig) return;

  try {
    const dbPool = await initDbConnection(userId, dbConfig);
    const wsConn = initWsConnection(userId, wsConfig, dbPool);
    userConnections.set(userId, { dbConn: dbPool, wsConn });
  } catch (err) {
    console.error(`初始化用户 ${userId} 连接失败:`, err);
  }
}

// 应用启动时加载所有用户的连接
async function loadAllUserConnections(UserModel) {
  const users = await UserModel.find({ dbConfig: { $exists: true }, wsConfig: { $exists: true } });
  for (const user of users) {
    await initUserConnections(user);
  }
}

// 销毁用户连接（比如用户删除配置时）
function destroyUserConnections(userId) {
  const connections = userConnections.get(userId);
  if (connections) {
    connections.dbConn.close();
    connections.wsConn.close();
    userConnections.delete(userId);
    console.log(`用户 ${userId} 连接已销毁`);
  }
}

module.exports = {
  userConnections,
  initUserConnections,
  loadAllUserConnections,
  destroyUserConnections
};

3. 修改原有模块为可复用函数 #

修改dbConn.js #

从单例模式改为传入配置返回连接的函数：

// dbConn.js
const sql = require('mssql');

module.exports = {
  getDbConnection: async (config) => {
    try {
      return await sql.connect({
        ...config,
        trustServerCertificate: true
      });
    } catch (err) {
      console.error('数据库连接失败:', err);
      throw err;
    }
  }
};

4. 改造App.js的初始化逻辑 #

在应用启动时加载所有用户的连接，同时处理用户提交配置后的连接初始化：

// App.js
// ...原有代码...
const connectionManager = require('./connectionManager');
const User = require('./models/user');

// ...中间件和路由配置...

mongoose.connect(MONGODB_URI)
  .then(async (result) => {
    // 应用启动时加载所有用户的连接
    await connectionManager.loadAllUserConnections(User);
    app.listen(3000);
    console.log('应用启动，已加载所有用户连接');
  })
  .catch((err) => {
    console.log(err);
  });

// 示例：添加路由处理用户提交配置（需认证）
app.post('/user/config', csrfProtection, async (req, res) => {
  if (!req.session.isLoggedIn) return res.redirect('/login');
  
  const { dbConfig, wsConfig } = req.body;
  try {
    // 更新用户配置
    await User.findByIdAndUpdate(req.user._id, { dbConfig, wsConfig });
    // 销毁旧连接（如果存在）
    connectionManager.destroyUserConnections(req.user._id);
    // 初始化新连接
    await connectionManager.initUserConnections(await User.findById(req.user._id));
    res.redirect('/home?configUpdated=true');
  } catch (err) {
    console.error('更新用户配置失败:', err);
    req.flash('error', '配置更新失败');
    res.redirect('/home');
  }
});

5. 修改工具函数（custom.js） #

让工具函数接收数据库连接作为参数，操作对应用户的数据库：

// util/custom.js
module.exports = {
  async syncMessages(messageId, dbPool) {
    try {
      const request = dbPool.request();
      // 使用传入的dbPool执行SQL操作
      const result = await request.query(`SELECT * FROM Messages WHERE Id = '${messageId}'`);
      // 后续逻辑...
    } catch (err) {
      console.error('同步消息失败:', err);
    }
  },
  // 同理修改syncHouses、syncMap等函数
  async syncHouses(dbPool) {
    // ...业务逻辑...
  }
};

关键注意事项 #

• 连接隔离：每个用户的数据库和WebSocket连接完全独立，避免数据交叉污染
• 资源管理：使用Map存储连接，在用户删除配置或账号时及时销毁连接，防止内存泄漏
• 重连机制：针对单个用户的连接实现重连，不影响其他用户
• 持久化：用户配置存储在MongoDB，应用重启后自动恢复所有连接，满足"未登录时连接保持活跃"的需求

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