🎯 学习目标

学完本章，你将能够：

• 理解 WebSocket 协议，能用 ws 库实现双向通信
• 使用 SSE 实现服务端向客户端的单向推送
• 使用 Socket.io 构建带房间、命名空间的实时应用
• 实现在线状态系统和断线重连机制
• 将认证集成到实时通信中

你有没有想过

微信聊天消息为什么能"秒到"？股票价格为什么能实时跳动？

这些都不是前端不停刷新请求来的。它们用的是 WebSocket——一种让服务器能主动向客户端推送数据的协议。

传统的 HTTP 是"你问我答"，WebSocket 是"随时聊"。这一章你将学会：WebSocket 原理、Socket.io 框架、实现一个完整的聊天室。

Part 1: WebSocket

从 HTTP 升级到全双工通信

WebSocket 是全双工通信协议，和 HTTP 的"请求-响应"模式不同，它建立一次连接后，双方可以随时互发消息。这一部分我们先掌握 WebSocket 的原理和原生用法，再对比 SSE——另一种服务端推送方案，理解什么场景该选什么技术。

HTTP vs WebSocket

通信模式完全不同

握手过程

WebSocket 连接从一个 HTTP 升级请求开始

客户端发起 Upgrade: websocket 请求，服务器返回 101 Switching Protocols，之后连接就从 HTTP 升级为 WebSocket 协议。

ws 库使用

Node.js 生态中最流行的原生 WebSocket 库

$ npm install ws
$ npm install -D @types/ws

import { WebSocketServer, WebSocket } from 'ws';
import http from 'http';

const server = http.createServer();
const wss = new WebSocketServer({ server, path: '/ws' });

const clients = new Map<string, { ws: WebSocket; userId?: string }>();

wss.on('connection', (ws) => {
    const clientId = Math.random().toString(36).substring(2, 10);
    clients.set(clientId, { ws });

    ws.on('message', (raw) => {
        const message = JSON.parse(raw.toString());
        switch (message.type) {
            case 'ping':
                ws.send(JSON.stringify({ type: 'pong' }));
                break;
            case 'broadcast':
                const data = JSON.stringify({ type: 'message', data: message.data });
                clients.forEach((client) => {
                    if (client.ws.readyState === WebSocket.OPEN) client.ws.send(data);
                });
                break;
        }
    });

    ws.on('close', () => clients.delete(clientId));
});

server.listen(3000);

浏览器客户端

浏览器原生支持 WebSocket API

const ws = new WebSocket('ws://localhost:3000/ws');

ws.addEventListener('open', () => {
    console.log('已连接');
    ws.send(JSON.stringify({ type: 'broadcast', data: { content: '你好' } }));
});

ws.addEventListener('message', (event) => {
    const message = JSON.parse(event.data);
    console.log('收到:', message);
});

ws.addEventListener('close', (event) => {
    console.log('断开:', event.code);
});

SSE (Server-Sent Events)

不是所有场景都需要双向通信

SSE 服务端实现

用 Express 搭建 SSE 端点

// Express SSE 端点
import express from 'express';

const app = express();

app.get('/events', (req, res) => {
    // 设置 SSE 必需的响应头
    res.writeHead(200, {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
    });

    // 定时推送数据
    const interval = setInterval(() => {
        const data = JSON.stringify({
            time: new Date().toISOString(),
            message: `服务时间: ${new Date().toLocaleTimeString()}`,
        });
        // SSE 格式: "data: ...\n\n"
        res.write(`data: ${data}\n\n`);
    }, 1000);

    // 客户端断开时清理
    req.on('close', () => {
        clearInterval(interval);
        res.end();
    });
});

app.listen(3000);

SSE 的核心就是三个响应头：Content-Type: text/event-stream 告诉浏览器这是 SSE 流，Cache-Control: no-cache 禁止缓存，Connection: keep-alive 保持长连接。数据格式固定为 data: 内容\n\n。

SSE 客户端：EventSource

浏览器原生 API，开箱即用

// 前端代码
const evtSource = new EventSource('http://localhost:3000/events');

evtSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log('收到推送:', data.message);
};

// 监听自定义事件名
evtSource.addEventListener('notification', (event) => {
    const notification = JSON.parse(event.data);
    showToast(notification.title, notification.body);
});

// 连接出错时浏览器会自动重连
evtSource.onerror = (event) => {
    console.log('SSE 连接中断，浏览器将自动重连...');
};

// 主动关闭
// evtSource.close();

// 服务端推送自定义事件名
// res.write(`event: notification\n`);
// res.write(`data: ${JSON.stringify({ title: '新消息', body: '你好' })}\n\n`);

EventSource 的 API 比 WebSocket 更简单：不用手动处理重连，浏览器断线后自动重试。服务端用 event: 字段可以推送具名事件，客户端用 addEventListener 按名监听。

Part 2: Socket.io 进阶

生产级实时通信框架

Socket.io 是建立在 WebSocket 之上的框架，自动处理重连、房间、广播，浏览器不支持 WebSocket 时自动降级到轮询。这一部分我们将学习如何把 Socket.io 挂载到 Express 服务器上、用 Redis 适配器支持多实例部署、实现在线状态系统、处理断线重连，以及集成 JWT 认证。

事件类型定义

前后端共享类型，保证事件名和载荷类型一致

$ npm install socket.io
$ npm install socket.io-client  # 前端

// src/types/socket.events.ts
export interface ServerToClientEvents {
    'chat:message': (message: ChatMessage) => void;
    'chat:history': (messages: ChatMessage[]) => void;
    'room:joined': (data: { room: string; users: string[] }) => void;
    'user:online': (userId: string) => void;
    'user:offline': (userId: string) => void;
}

export interface ClientToServerEvents {
    'chat:message': (data: { room: string; content: string }) => void;
    'room:join': (room: string) => void;
    'room:leave': (room: string) => void;
}

export interface ChatMessage {
    id: string;
    userId: string;
    username: string;
    content: string;
    room: string;
    timestamp: number;
}

export interface SocketData {
    userId: string;
    username: string;
    rooms: string[];
}

Socket.io 服务端

类型安全的 Socket.io 服务器配置

import { Server } from 'socket.io';
import { createServer } from 'http';

const httpServer = createServer();
const io = new Server<ClientToServerEvents, ServerToClientEvents, {}, SocketData>(httpServer, {
    cors: { origin: 'http://localhost:5173' },
    pingInterval: 10000,
    pingTimeout: 5000,
});

io.use((socket, next) => {
    const token = socket.handshake.auth.token;
    if (!token) return next(new Error('缺少令牌'));
    // 验证 token...
    socket.data.userId = 'user_123';
    socket.data.username = 'Alice';
    socket.data.rooms = [];
    next();
});

io.on('connection', (socket) => {
    console.log(`${socket.data.username} 已连接`);

    socket.on('room:join', (room) => {
        socket.join(room);
        io.to(room).emit('room:joined', { room, users: [] });
    });

    socket.on('chat:message', (data) => {
        io.to(data.room).emit('chat:message', {
            id: Date.now().toString(36),
            userId: socket.data.userId,
            username: socket.data.username,
            content: data.content,
            room: data.room,
            timestamp: Date.now(),
        });
    });

    socket.on('disconnect', () => {
        io.emit('user:offline', socket.data.userId);
    });
});

Socket.io 客户端

前端连接 Socket.io 服务器

// 前端代码
import { io, Socket } from 'socket.io-client';

// token 从内存（状态管理）读取，不要用 localStorage——会暴露给 XSS（见第 44 章）
const socket: Socket<ServerToClientEvents, ClientToServerEvents> = io(
    'http://localhost:3000',
    {
        autoConnect: false,
        auth: { token: getAccessToken() },
    }
);

socket.on('chat:message', (message) => {
    console.log(`${message.username}: ${message.content}`);
});

socket.connect();

// 发送消息
socket.emit('chat:message', { room: 'general', content: '你好' });

Socket.io + Express 集成

把 Socket.io 挂载到现有的 Express HTTP 服务器上

import express from 'express';
import { createServer } from 'http';
import { Server } from 'socket.io';
import session from 'express-session';

const app = express();
const httpServer = createServer(app);

// Socket.io 挂载到同一个 HTTP 服务器
const io = new Server(httpServer, {
    cors: { origin: 'http://localhost:5173', credentials: true },
});

// Express 中间件
app.use(express.json());
app.use(session({ secret: 'your-secret', resave: false, saveUninitialized: false }));

// REST API 路由正常工作
app.get('/api/messages', async (req, res) => {
    const messages = await MessageService.getRoomMessages('general');
    res.json(messages);
});

// Socket.io 共享 session
io.engine.use(session({ secret: 'your-secret', resave: false, saveUninitialized: false }));

io.on('connection', (socket) => {
    // 从握手请求中读取 session
    const session = socket.request.session;
    console.log('用户 session:', session.userId);
});

// 两者共用同一个端口
httpServer.listen(3000, () => {
    console.log('HTTP + WebSocket 服务运行在 3000 端口');
});

关键点：Socket.io 通过 createServer(app) 挂载到 Express 的底层 HTTP 服务器上，REST API 和 WebSocket 共用同一个端口。用 io.engine.use() 可以让 WebSocket 握手请求也走 Express 中间件，实现 session 共享。

Redis 适配器

多实例部署时，不同服务器上的 Socket.io 需要通信

$ npm install @socket.io/redis-adapter redis

import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';

const pubClient = createClient({ url: 'redis://localhost:6379' });
const subClient = pubClient.duplicate();

await Promise.all([pubClient.connect(), subClient.connect()]);

const io = new Server(httpServer, {
    cors: { origin: 'http://localhost:5173' },
});

// 挂载 Redis 适配器
io.adapter(createAdapter(pubClient, subClient));

// 现在 io.emit() 和 io.to(room).emit() 会跨实例广播
io.on('connection', (socket) => {
    console.log(`连接来自实例: ${process.pid}`);
});

单实例部署时 Socket.io 用内存就够了。一旦有多个实例（PM2 集群、Docker 副本、K8s Pod），消息广播就需要通过 Redis Pub/Sub 同步到所有实例。否则连接在服务器 A 的用户发消息，连接在服务器 B 的用户收不到。

在线状态系统

追踪谁在线，谁离开了

// 在线用户管理
const onlineUsers = new Map<string, { userId: string; username: string; rooms: string[] }>();

io.on('connection', (socket) => {
    const { userId, username } = socket.data;

    // 用户上线
    onlineUsers.set(socket.id, { userId, username, rooms: [] });
    io.emit('presence:online', { userId, username, onlineCount: onlineUsers.size });

    // 加入房间时更新
    socket.on('room:join', (room) => {
        socket.join(room);
        const user = onlineUsers.get(socket.id);
        if (user) {
            user.rooms.push(room);
            // 向房间内广播在线用户列表
            const roomUsers = [...onlineUsers.values()]
                .filter(u => u.rooms.includes(room));
            io.to(room).emit('room:users', roomUsers.map(u => ({
                userId: u.userId,
                username: u.username,
            })));
        }
    });

    // 用户下线
    socket.on('disconnect', () => {
        const user = onlineUsers.get(socket.id);
        onlineUsers.delete(socket.id);
        io.emit('presence:offline', {
            userId: user?.userId,
            onlineCount: onlineUsers.size,
        });
    });
});

在线状态系统的核心是维护一个 Map<socketId, userInfo>，在 connection 时添加、disconnect 时移除，并通过自定义事件广播上下线通知。多实例部署时，可以把状态存到 Redis 而不是内存。

断线重连与状态恢复

网络不稳定时，用户体验不能断

// 客户端：Socket.io 自动重连配置
const socket = io('http://localhost:3000', {
    reconnection: true,
    reconnectionAttempts: 10,     // 最多重试 10 次
    reconnectionDelay: 1000,      // 初始延迟 1s
    reconnectionDelayMax: 5000,   // 最大延迟 5s
    randomizationFactor: 0.5,     // 随机抖动因子
});

// 监听重连事件
socket.on('connect', () => {
    console.log('已连接，重新加入房间');
    // 重连后需要重新加入之前的房间
    currentRooms.forEach(room => socket.emit('room:join', room));
    // 请求离线期间的消息
    socket.emit('message:catchup', { lastTimestamp });
});

socket.on('disconnect', (reason) => {
    console.log('断开原因:', reason);
    // 'io server disconnect' — 服务端主动断开，需要手动重连
    if (reason === 'io server disconnect') {
        socket.connect();
    }
    // 其他原因 Socket.io 会自动重连
});

// 服务端：为断线重连的用户提供消息补发
socket.on('message:catchup', async ({ lastTimestamp }) => {
    const missedMessages = await MessageService.getMessagesAfter(lastTimestamp);
    socket.emit('chat:history', missedMessages);
});

Socket.io 内置了自动重连机制，支持指数退避策略。但重连成功后，需要客户端主动重新加入房间和请求离线期间的消息，这些逻辑需要你手动实现。

Socket.io 认证集成

用 JWT 保护 WebSocket 连接

import jwt from 'jsonwebtoken';

// Socket.io 认证中间件
io.use((socket, next) => {
    const token = socket.handshake.auth.token
        || socket.handshake.headers.authorization?.replace('Bearer ', '');

    if (!token) {
        return next(new Error('AUTH_REQUIRED'));
    }

    try {
        const decoded = jwt.verify(token, process.env.JWT_SECRET!);

        // 检查 token 是否即将过期（可选）
        const expiresIn = (decoded.exp * 1000) - Date.now();
        if (expiresIn < 0) {
            return next(new Error('TOKEN_EXPIRED'));
        }

        // 将用户信息挂载到 socket.data
        socket.data.userId = decoded.userId;
        socket.data.username = decoded.username;
        next();
    } catch (err) {
        // 区分不同错误类型
        if (err.name === 'TokenExpiredError') {
            return next(new Error('TOKEN_EXPIRED'));
        }
        return next(new Error('TOKEN_INVALID'));
    }
});

// 客户端：连接时传递 token（token 从内存读取，而非 localStorage）
const socket = io('http://localhost:3000', {
    auth: { token: getAccessToken() },
});

// 监听认证错误
socket.on('connect_error', (err) => {
    if (err.message === 'TOKEN_EXPIRED') {
        // 刷新 token 后重连
        refreshToken().then(newToken => {
            socket.auth.token = newToken;
            socket.connect();
        });
    }
});

WebSocket 握手没有 HTTP 请求体，token 通过 handshake.auth 或 headers.authorization 传递。验证失败时调用 next(new Error(...))，客户端会在 connect_error 事件中收到错误消息，据此决定是刷新 token 还是跳转到登录页。

Part 3: 聊天室实战

从零搭建一个完整聊天应用

把前面学的知识组合起来，做一个完整的聊天室应用。这个项目涵盖实时消息收发、房间隔离、历史消息分页、在线状态追踪、消息持久化等核心功能，是你掌握实时通信的综合性练习。

项目结构

清晰分层，实时通信和 REST API 共存

chat-app/
  src/
    server.ts           # 服务器入口
    websocket/
      socket-server.ts  # Socket.io 配置
      chat.handler.ts   # 聊天业务逻辑
    routes/
      chat.routes.ts    # REST API（历史消息）
    services/
      message.service.ts # 消息存储
    types/
      socket.events.ts  # 类型定义

消息服务

用 Prisma 持久化聊天消息

// prisma/schema.prisma
// model Message {
//   id        String   @id @default(cuid())
//   userId    String
//   username  String
//   content   String   @db.Text
//   room      String   @default("general")
//   createdAt DateTime @default(now())
//   @@index([room, createdAt])
// }

// src/services/message.service.ts
export class MessageService {
    static async saveMessage(data: { userId: string; username: string; content: string; room: string }) {
        return prisma.message.create({ data });
    }

    static async getRoomMessages(room: string, limit = 50) {
        const messages = await prisma.message.findMany({
            where: { room },
            orderBy: { createdAt: 'desc' },
            take: limit,
        });
        return messages.reverse();
    }
}

功能清单

一个完整聊天室需要的七大功能

常见问题

问题一：客户端连不上 WebSocket

原因：CORS 配置不对，或服务器端口没监听。

修复：在 Socket.io 配置中设置正确的 cors.origin，确认服务器在监听正确的端口。

问题二：消息发送了但对方收不到

原因：发送者和接收者不在同一个 Socket.io 房间。

修复：确认双方都执行了 socket.join(room)，用 io.to(room).emit(...) 而不是 socket.emit(...)。

问题三：频繁断线重连

原因：网络不稳定，或心跳超时设置太短。

修复：增大 pingInterval 和 pingTimeout 的值，或在客户端设置 reconnectionAttempts。

试一试

// 提示：ws.on('message', ...) 中调用 ws.send()
// 你的代码：

wss.on('connection', (ws) => {
    ws.on('message', (data) => {
        ws.send(data); // 原样返回
    });
});

// 提示：监听 chat:typing 事件，用 socket.to(room).emit 广播
// 你的代码：

socket.on('chat:typing', (data) => {
    socket.to(data.room).emit('chat:typing', {
        room: data.room,
        isTyping: data.isTyping,
        username: socket.data.username,
    });
});

📝 要点总结

本章核心知识点