Node.js中如何正确抽离expressWs模块逻辑解决WebSocket已关闭报错

核心错误点 #

• 重复注册WebSocket路由：主文件中调用app.ws('', websocketController(app))时，会在每次根路径有WS连接请求时，重复给app实例注册/echo路由，逻辑完全混乱，且根路径产生的WS连接实例未被处理，超时自动关闭后就会触发状态错误。
• 状态变量重复定义：主文件和controller中分别定义了独立的history、clients变量，聊天记录和在线用户列表完全隔离，无法正常同步。
• 群发未做连接状态校验：遍历客户端群发消息时，没有判断当前客户端是否处于活跃连接状态，直接调用send就会抛出连接已关闭的错误。
• WS连接逻辑嵌套错误：controller中把/echo的路由注册逻辑写在了根路径WS连接的回调函数里，路由注册时机完全错误。

修正后代码 #

1. expressWs.js（主服务文件） #

require('dotenv').config() 
const sequelize = require("./db")
const fileUpload = require("express-fileupload")
const cors  = require("cors")
const router = require("./routes/index")
const errorHandler = require("./middleware/ErrorHandlingMiddleware")
const path = require("path")
const fs = require('fs')
const websocketController = require('./websocket/websocketController')
var express = require('express');
var expressWs = require('express-ws');
const { send } = require('process')
var expressWs = expressWs(express());
var app = expressWs.app;

app.use(express.static('public'));
const PORT = process.env.PORT

app.use(cors({
  origin: '*'
}))
app.use(express.json())
app.use(fileUpload({}))
app.use(express.static(path.resolve(__dirname, "static")))
app.use('/api',router)
app.use(errorHandler)

// 全局共享状态，不要在controller中重复定义
let wsState = {
  history: [],
  clients: []
}
// 直接注册/echo路由，传入app实例和全局状态
app.ws('/echo', websocketController(app, wsState))

const start = async () => {
    try {
        await sequelize.authenticate()
        await sequelize.sync()
        app.listen(PORT);
        console.log(`Server started on port ${PORT}`)
    } catch (e){
        console.log(e)
    }
}

start()

2. websocketController.js（业务逻辑文件） #

module.exports = function websocketController(app, wsState){
  const { history, clients } = wsState
  return function (ws, req) {
    const aWss = app.getWss('/echo')
    let currentClient = ''
    console.log('Socket Connected');
    // 新连接发送历史消息
    ws.send(JSON.stringify(history))
    console.log('history',history)

    ws.on('message', msg => {
      msg = JSON.parse(msg)
      currentClient =  msg.username.slice()
      
      if(msg.event === 'message'){
        history.push(msg)
        // 保留最近100条记录，原代码slice没有赋值，完全不生效
        history = history.slice(-100)
      }
      if(msg.event === 'connection'){
        clients.push(currentClient)
      }
      console.log('clients',clients)
      
      aWss.clients.forEach(client => {
        // 发送前先判断连接状态，避免报错
        if (client.readyState === 1) {
          client.send(JSON.stringify(msg))
        }
      })
    })

    ws.on('close', () => {
      console.log('WebSocket was closed')
      console.log('currentClient',currentClient)
      const index = clients.indexOf(currentClient)
      if (index > -1) {
        clients.splice(index, 1)
      }
      console.log('clients after filter',clients)
      currentClient = undefined
    })
  }
}

修改说明 #

• 移除了根路径的无效WS路由注册，直接把/echo路由的注册逻辑放在主文件中，避免重复注册。
• 把全局状态history、clients统一放在主文件管理，通过参数传入controller，保证状态全局同步。
• 群发消息前增加client.readyState === 1的判断，只有处于OPEN状态的连接才会发送消息，避免抛出关闭状态错误。
• 修复了历史记录截取逻辑，history.slice(-100)需要重新赋值给history才会生效。
• currentClient改为每个连接私有的变量，避免多连接并发时变量覆盖的问题。

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