是否存在标准WebSocket编写方式可统一多端实现逻辑？

完全可以通过API对齐封装的方式抹平不同语言、不同框架的写法差异，让服务端的连接实例接口完全对齐浏览器原生WebSocket的规范，不需要单独学习不同框架的差异化语法。

核心实现思路 #

所有WebSocket交互的核心逻辑本质是一致的：

• 服务端负责监听端口、处理新连接
• 每个连接实例具备4个核心生命周期事件：open（连接建立）、message（收到消息）、close（连接断开）、error（连接异常）
• 每个连接实例具备1个核心发送方法：send()
  我们只需要对不同语言的底层WebSocket库做薄封装，把原生API统一映射成上面的标准接口即可，封装逻辑仅需实现一次，业务开发全程无需感知底层差异。

具体落地示例 #

封装后Node.js服务端代码（对齐浏览器API） #

import { WebSocketServer } from 'ws';

// 底层适配封装
const createWSServer = (port) => {
  const wss = new WebSocketServer({ port });
  wss.on('connection', (ws) => {
    // 事件转发对齐浏览器标准
    ws.dispatchEvent(new Event('open'));
    ws.on('message', (data) => {
      ws.dispatchEvent(new MessageEvent('message', { data }));
    });
    ws.on('close', () => ws.dispatchEvent(new Event('close')));
    ws.on('error', (err) => ws.dispatchEvent(new ErrorEvent('error', { error: err })));
  });
  return wss;
}

// 业务代码写法和浏览器完全一致
const wss = createWSServer(8080);
wss.on('connection', (socket) => {
  socket.addEventListener('open', (event) => {
    socket.send('something');
  });
  socket.addEventListener('message', (event) => {
    console.log('received: %s', event.data);
  });
});

封装后Python服务端代码（对齐浏览器API） #

import asyncio
import websockets
from dataclasses import dataclass

# 底层适配封装
@dataclass
class MessageEvent:
    type: str
    data: str

class WrappedWebSocket:
    def __init__(self, ws):
        self.ws = ws
        self._listeners = {"open": [], "message": [], "close": [], "error": []}
    
    def add_event_listener(self, event_type, callback):
        self._listeners[event_type].append(callback)
    
    async def send(self, data):
        await self.ws.send(data)
    
    async def run(self):
        for cb in self._listeners["open"]:
            cb(None)
        try:
            async for message in self.ws:
                event = MessageEvent(type="message", data=message)
                for cb in self._listeners["message"]:
                    cb(event)
        except Exception as e:
            for cb in self._listeners["error"]:
                cb(e)
        finally:
            for cb in self._listeners["close"]:
                cb(None)

async def create_ws_server(port):
    async def handler(ws, path):
        wrapped = WrappedWebSocket(ws)
        await wrapped.run()
    return await websockets.serve(handler, "localhost", port)

# 业务代码写法和浏览器高度对齐
async def main():
    server = await create_ws_server(8000)
    def handle_connection(socket):
        def on_open(e):
            socket.send("something")
        def on_message(e):
            print(f"received: {e.data}")
        socket.add_event_listener("open", on_open)
        socket.add_event_listener("message", on_message)
    async with server:
        await asyncio.Future()

asyncio.run(main())

额外优化方案 #

如果需要进一步降低适配成本，可以采用代码生成工具，根据统一的接口定义自动生成不同语言的适配层代码，完全不需要手动编写封装逻辑。
经过以上处理后，业务开发人员只需要掌握浏览器原生WebSocket的API写法，就可以同时编写Python、Node.js服务端和浏览器客户端的WebSocket逻辑，不需要记忆不同框架的差异化用法。

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