WebSocket vs 其他实时方案
实时通信方案对比：
┌─────────────────────────────────────────┐
│ WebSocket │
│ ✓ 全双工通信，低延迟 │
│ ✓ 持久连接，减少握手开销 │
│ ✓ 适合高频双向交互（聊天、游戏） │
│ ✗ 需要维护连接状态 │
│ ✗ 扩展性需要考虑连接亲和性 │
│ │
│ Server-Sent Events (SSE) │
│ ✓ 基于HTTP，简单实现 │
│ ✓ 自动重连，易于调试 │
│ ✓ 适合单向推送（通知、股票行情） │
│ ✗ 只能服务器到客户端 │
│ ✗ 连接数受限（HTTP/1.1每域6个） │
│ │
│ 长轮询（Long Polling） │
│ ✓ 兼容性好，无需特殊支持 │
│ ✓ 实现简单 │
│ ✗ 延迟较高 │
│ ✗ 服务器资源消耗大 │
│ │
│ 选择建议： │
│ - 双向高频交互 → WebSocket │
│ - 单向实时推送 → SSE │
│ - 兼容性要求高 → 长轮询 │
└─────────────────────────────────────────┘
SSE 服务端推送事件详解
在构建实时通信系统时，并非所有场景都需要 WebSocket 的全双工能力。对于纯服务端到客户端的单向数据流，Server-Sent Events（SSE） 是一种更轻量、更符合 REST 风格的技术选择。SSE 基于标准 HTTP 协议，天然支持自动重连（Auto-Reconnect）和断点续传（resume via Last-Event-ID），并且能够穿透大多数企业防火墙和代理服务器。与 WebSocket 相比，SSE 不需要额外的握手开销，也不需要处理复杂的二进制帧解析逻辑。对于股票行情推送、新闻实时通知、后台任务进度反馈等场景，SSE 往往是更稳妥的方案。当系统需要扩展而非 WebSocket 独占的推送方案时，也可以参考 https://plumephp.com/graphql-subscriptions-realtime-updates/ 中关于订阅机制的实现对比。
// sse-server.js — 基于 Express 的 SSE 服务端实现
const express = require('express');
const { v4: uuidv4 } = require('uuid');
const app = express();
const clients = new Map();
app.get('/events', (req, res) => {
const lastEventId = req.headers['last-event-id'] || '0';
const clientId = uuidv4();
res.setHeader('Content-Type', 'text/event-stream');
res.setHeader('Cache-Control', 'no-cache');
res.setHeader('Connection', 'keep-alive');
res.setHeader('X-Accel-Buffering', 'no');
res.flushHeaders();
clients.set(clientId, res);
console.log(`SSE client connected: ${clientId}, lastEventId=${lastEventId}`);
// 推送断点续传期间累积的离线消息
const missedMessages = getMissedMessages(Number(lastEventId));
missedMessages.forEach(msg => {
res.write(`id: ${msg.id}\n`);
res.write(`event: ${msg.event}\n`);
res.write(`data: ${JSON.stringify(msg.data)}\n\n`);
});
req.on('close', () => {
clients.delete(clientId);
console.log(`SSE client disconnected: ${clientId}`);
});
});
// 广播推送函数（支持按频道过滤）
function broadcast(event, data, channel = 'all') {
const id = Date.now();
const payload = `id: ${id}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
clients.forEach((res) => {
res.write(payload);
});
}
setInterval(() => {
broadcast('heartbeat', { ts: Date.now() });
}, 30000);
app.listen(3001, () => {
console.log('SSE server running on port 3001');
});
SSE 的 EventSource 客户端接口天然支持自动重连，无需手动实现心跳逻辑。通过设置 id 字段，浏览器在断线恢复时会自动带上 Last-Event-ID 请求头，服务端据此推送断线期间遗漏的消息，从而实现可靠的有序推送。
WebSocket安全威胁与防护
WebSocket 协议虽然功能强大，但也引入了新的攻击面。在生产环境中部署 WebSocket 服务时，必须防范 CSWSH（Cross-Site WebSocket Hijacking）、帧注入、资源耗尽攻击等风险。CSWSH 类似于 CSRF 攻击，攻击者诱导用户的浏览器发起 WebSocket 连接到目标站点，并携带用户的认证 Cookie，从而达到冒充用户的目的。由于 WebSocket 握手使用的是 HTTP Upgrade 请求，传统的 CSRF Token 防护机制在此阶段并不直接适用。因此，服务端必须在 Sec-WebSocket-Key 之外，额外校验 Origin 头部或要求握手阶段携带不可预测的 Token。同时，对于大规模部署，还需要在网关层实现速率限制（Rate Limiting）和消息大小检查，防止恶意客户端通过超大帧 or 高频消息耗尽服务器资源。关于现代 API 网关的防护策略，可以进一步了解 https://plumephp.com/api-gateway-ratelimiting-circuitbreaker-practice/ 中针对 WebSocket 流量限流的具体方案。
// security/middleware.go — WebSocket 安全防护中间件
package security
import (
"net/http"
"strings"
"time"
"github.com/gorilla/websocket"
"golang.org/x/time/rate"
)
var upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool {
origin := r.Header.Get("Origin")
allowed := []string{"https://app.example.com", "https://admin.example.com"}
for _, o := range allowed {
if strings.EqualFold(origin, o) {
return true
}
}
return false
},
}
// RateLimiter 基于令牌桶的每连接限流器
type RateLimiter struct {
limiter *rate.Limiter
}
func NewRateLimiter(rps rate.Limit, burst int) *RateLimiter {
return &RateLimiter{limiter: rate.NewLimiter(rps, burst)}
}
func (rl *RateLimiter) Allow() bool {
return rl.limiter.Allow()
}
// SecureUpgrade 在升级前执行多层安全检查
func SecureUpgrade(w http.ResponseWriter, r *http.Request, checkToken func(string) error) (*websocket.Conn, error) {
// 1. 验证握手 Token（防御 CSWSH）
token := r.URL.Query().Get("token")
if token == "" || checkToken(token) != nil {
http.Error(w, "Invalid handshake token", http.StatusForbidden)
return nil, websocket.ErrBadHandshake
}
// 2. 帧大小限制已在 Upgrader 层面通过 CheckOrigin 和后续 ReadLimit 控制
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
return nil, err
}
// 3. 设置读取上限（防御超大帧攻击）
conn.SetReadLimit(65536)
conn.SetReadDeadline(time.Now().Add(120 * time.Second))
return conn, nil
}
在生产环境中，建议将 WebSocket 服务置于 WAF 或 API 网关之后，利用网关的统一策略管理 Origin 白名单、连接数上限和首包大小限制，从而在边缘层过滤掉绝大多数恶意流量。
微服务架构中的WebSocket网关
在微服务架构下，WebSocket 连接管理如果分散在各个业务服务中，会导致重复建设、认证逻辑不统一、服务间消息路由混乱等问题。引入独立的 WebSocket 网关层 是解决这些痛点的关键举措。WebSocket 网关承担统一入口职责：负责长连接的保持与心跳管理，对客户端请求进行 JWT 认证与权限校验，并根据消息中的路由字段将业务消息转发到对应的后端服务（如聊天服务、通知服务、游戏房间服务等）。网关本身无状态，水平扩展时通过 Redis 等共享存储同步会话信息。在服务发现层面，网关可以与 Consul、etcd 或 ZooKeeper 集成，动态感知后端服务的健康状态，实现流量的智能调度。若对分布式协调组件感兴趣，可深入阅读 https://plumephp.com/zookeeper-coordination/ 中关于服务注册与选主机制的内容。
# websocket-gateway-config.yaml — 网关路由与认证配置
apiVersion: v1
kind: ConfigMap
metadata:
name: websocket-gateway-config
data:
gateway.yaml: |
server:
port: 8080
maxConnectionsPerIP: 5000
heartbeatInterval: 25s
heartbeatTimeout: 60s
auth:
type: jwt
secretEnv: JWT_SECRET
tokenHeader: Sec-WebSocket-Protocol
tokenQueryParam: token
routing:
rules:
- prefix: "chat."
service: chat-service
namespace: realtime
port: 8081
- prefix: "notify."
service: notification-service
namespace: realtime
port: 8082
- prefix: "game."
service: game-room-service
namespace: realtime
port: 8083
discovery:
type: consul
address: consul-server.default.svc.cluster.local:8500
healthCheckInterval: 10s
metrics:
enabled: true
path: /metrics
port: 9090
WebSocket 网关的核心设计原则是连接与业务分离。网关只负责维持长连接和轻量级消息路由，所有业务状态计算交给后端微服务处理。这种分层架构使得前端无需关心后端服务的部署拓扑，同时也方便对 WebSocket 流量进行统一监控和限流。
消息队列与WebSocket的集成
在高并发的实时系统中，WebSocket 服务器直接处理所有业务逻辑会导致单点瓶颈和耦合度过高。更优雅的架构是将 WebSocket 定位为最后一公里推送层，而将可靠的消息分发交由专业的消息队列（如 Kafka、NATS、RabbitMQ）承担。消息队列作为后端事件总线，负责削峰填谷、消息持久化以及跨服务的事件广播；WebSocket 服务器则消费队列中的事件，并将其推送给在线客户端。这种模式不仅解耦了消息生产者与消费者，还使得离线消息回放、多机房同步、灰度发布等高级能力变得易于实现。例如，当用户下单成功后，订单服务将事件发布到 Kafka 的 order.completed Topic，WebSocket 消费组拉取消息后，查询哪些用户当前在线并推送通知。关于事件驱动架构的更多设计模式，请参阅 https://plumephp.com/event-driven-architecture-domain-events/ 中领域事件与消息总线的章节。
// kafka-websocket-bridge.js — Kafka 消费者桥接 WebSocket 推送
const { Kafka } = require('kafkajs');
const { Server } = require('socket.io');
const kafka = new Kafka({
clientId: 'websocket-bridge',
brokers: ['kafka-1:9092', 'kafka-2:9092', 'kafka-3:9092']
});
const consumer = kafka.consumer({ groupId: 'websocket-push-group' });
async function startBridge(io) {
await consumer.connect();
// 订阅多个业务 Topic
await consumer.subscribe({ topics: ['order.completed', 'notify.broadcast', 'chat.message'] });
await consumer.run({
eachMessage: async ({ topic, partition, message }) => {
const event = JSON.parse(message.value.toString());
const { userId, roomId, payload, type } = event;
switch (topic) {
case 'order.completed':
// 推送给指定用户（个人通知）
io.to(`user:${userId}`).emit('order:completed', payload);
break;
case 'notify.broadcast':
// 全站广播
io.emit('broadcast', payload);
break;
case 'chat.message':
// 推送给房间所有成员
if (roomId) {
io.to(`room:${roomId}`).emit('message:new', payload);
}
break;
}
console.log(`[Bridge] Forwarded ${topic} to user=${userId} room=${roomId}`);
}
});
}
// 优雅关闭：先停止消费再断开连接
async function gracefulShutdown() {
await consumer.disconnect();
console.log('Kafka bridge shut down gracefully');
}
module.exports = { startBridge, gracefulShutdown };
使用消息队列作为中间层还有一个显著优势：水平扩展的灵活性。当 WebSocket 服务器集群需要扩容时，只需增加消费者实例即可自动参与消息分发；当需要缩容时，Kafka 的再平衡机制会将分区重新分配，保证消息不丢失。这种架构使得 WebSocket 层成为无状态的推送终端，任何节点的故障都不影响整体服务的可用性。
百万级并发性能调优
当 WebSocket 服务需要支撑百万级并发连接时，仅靠代码层面的优化远远不够，必须从操作系统内核、网络栈、进程模型多个维度进行系统性调优。在 Linux 环境下，文件描述符限制是首要瓶颈。每个 WebSocket 连接至少占用一个文件描述符，因此必须通过调整 ulimit -n 以及 /etc/security/limits.conf 确保进程可打开的文件数达到百万级别。其次，启用 SO_REUSEPORT 选项可以让多个监听进程绑定到同一个端口，由内核在传输层进行负载均衡，避免单 Listen 进程成为瓶颈。此外，还需要调整 TCP 内核参数，如调大 net.ipv4.tcp_max_syn_backlog、net.core.somaxconn 和 net.netfilter.nf_conntrack_max，以应对突发的大量新建连接。Go 语言的 goroutine 模型天然适合高并发场景，但仍需关注 goroutine 的内存占用，通过设置合理的读写缓冲区大小和复用对象池来降低 GC 压力。对于需要异构语言服务协同的场景，Kafka 流处理是不可或缺的扩音器，可参考 https://plumephp.com/kafka-connect/ 中关于高吞吐消息管道的调优经验。
// tuning/server.go — 百万级并发调优配置
package main
import (
"net"
"syscall"
"github.com/gorilla/websocket"
)
// reusePortListen 使用 SO_REUSEPORT 支持多进程复用同一端口
func reusePortListen(network, address string) (net.Listener, error) {
lc := net.ListenConfig{
Control: func(network, address string, c syscall.RawConn) error {
var err error
c.Control(func(fd uintptr) {
// 启用端口复用，允许多个进程监听同一地址
err = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEPORT, 1)
})
return err
},
}
return lc.Listen(nil, network, address)
}
func main() {
listener, err := reusePortListen("tcp", ":8080")
if err != nil {
panic(err)
}
upgrader := websocket.Upgrader{
ReadBufferSize: 2048,
WriteBufferSize: 2048,
// 启用压缩以减少网络传输量（CPU 换带宽）
EnableCompression: true,
}
// 每连接尽量复用缓冲区，减少内存分配
upgrader.ReadBufferPool = &sync.Pool{
New: func() interface{} {
return make([]byte, 2048)
},
}
// ... 注册 handler 并启动服务
}
除了代码和内核调优，连接数监控也是百万级并发运维的必备手段。建议通过 Prometheus 采集 websocket_connected_clients、go_goroutines、process_open_fds 等指标，并配合 Grafana 设置告警阈值。当连接数接近 ulimit 上限或内存占用超过预设阈值时，及时触发自动扩容，确保服务在流量洪峰下依然稳健运行。
WebSocket服务器实现
Node.js + Socket.IO
// server.js
const express = require('express');
const http = require('http');
const { Server } = require('socket.io');
const redis = require('redis');
const jwt = require('jsonwebtoken');
const app = express();
const server = http.createServer(app);
// Socket.IO配置
const io = new Server(server, {
cors: {
origin: process.env.CLIENT_URL,
methods: ['GET', 'POST']
},
pingTimeout: 60000,
pingInterval: 25000
});
// Redis适配器（多进程支持）
const pubClient = redis.createClient({ url: process.env.REDIS_URL });
const subClient = pubClient.duplicate();
io.adapter(createAdapter(pubClient, subClient));
// 连接认证中间件
io.use(async (socket, next) => {
try {
const token = socket.handshake.auth.token;
if (!token) {
return next(new Error('Authentication error'));
}
const decoded = jwt.verify(token, process.env.JWT_SECRET);
socket.userId = decoded.userId;
socket.username = decoded.username;
next();
} catch (err) {
next(new Error('Authentication error'));
}
});
// 连接管理
const userSockets = new Map(); // userId -> Set<socketId>
io.on('connection', (socket) => {
console.log(`User ${socket.userId} connected: ${socket.id}`);
// 记录用户的socket连接
if (!userSockets.has(socket.userId)) {
userSockets.set(socket.userId, new Set());
}
userSockets.get(socket.userId).add(socket.id);
// 加入用户的个人房间
socket.join(`user:${socket.userId}`);
// 心跳检测
socket.on('pong', () => {
socket.isAlive = true;
});
// 加入聊天室
socket.on('join:room', async (roomId) => {
try {
// 验证用户是否有权限加入
const hasAccess = await checkRoomAccess(socket.userId, roomId);
if (!hasAccess) {
socket.emit('error', { message: 'Access denied' });
return;
}
socket.join(`room:${roomId}`);
socket.emit('room:joined', { roomId });
// 通知房间内其他用户
socket.to(`room:${roomId}`).emit('user:joined', {
userId: socket.userId,
username: socket.username
});
} catch (err) {
socket.emit('error', { message: 'Failed to join room' });
}
});
// 发送消息
socket.on('message:send', async (data) => {
try {
const { roomId, content, type = 'text' } = data;
// 创建消息记录
const message = await createMessage({
roomId,
userId: socket.userId,
content,
type,
timestamp: Date.now()
});
// 广播给房间内所有用户
io.to(`room:${roomId}`).emit('message:new', {
id: message.id,
roomId,
userId: socket.userId,
username: socket.username,
content,
type,
timestamp: message.timestamp
});
// 发送确认给发送者
socket.emit('message:sent', { messageId: message.id });
} catch (err) {
socket.emit('error', { message: 'Failed to send message' });
}
});
// 输入状态
socket.on('typing:start', (roomId) => {
socket.to(`room:${roomId}`).emit('user:typing', {
userId: socket.userId,
username: socket.username
});
});
socket.on('typing:stop', (roomId) => {
socket.to(`room:${roomId}`).emit('user:stop-typing', {
userId: socket.userId
});
});
// 断开连接
socket.on('disconnect', (reason) => {
console.log(`User ${socket.userId} disconnected: ${reason}`);
// 清理用户socket记录
const sockets = userSockets.get(socket.userId);
if (sockets) {
sockets.delete(socket.id);
if (sockets.size === 0) {
userSockets.delete(socket.userId);
}
}
// 通知相关房间
socket.rooms.forEach(room => {
if (room.startsWith('room:')) {
socket.to(room).emit('user:left', {
userId: socket.userId,
username: socket.username
});
}
});
});
});
// 心跳检测定时器
const heartbeatInterval = setInterval(() => {
io.sockets.sockets.forEach((socket) => {
if (socket.isAlive === false) {
return socket.disconnect(true);
}
socket.isAlive = false;
socket.emit('ping');
});
}, 30000);
// 清理
server.on('close', () => {
clearInterval(heartbeatInterval);
pubClient.quit();
subClient.quit();
});
server.listen(3000, () => {
console.log('WebSocket server running on port 3000');
});
Go实现（高并发场景）
// main.go
package main
import (
"context"
"encoding/json"
"log"
"net/http"
"sync"
"time"
"github.com/gorilla/websocket"
"github.com/redis/go-redis/v9"
)
var upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool {
return true // 生产环境需要验证origin
},
}
type Message struct {
Type string `json:"type"`
RoomID string `json:"roomId,omitempty"`
Content interface{} `json:"content"`
UserID string `json:"userId"`
Time int64 `json:"time"`
}
type Client struct {
ID string
UserID string
Conn *websocket.Conn
Send chan []byte
Rooms map[string]bool
mu sync.RWMutex
}
type Hub struct {
Clients map[string]*Client
Rooms map[string]map[string]*Client
Register chan *Client
Unregister chan *Client
Broadcast chan *BroadcastMessage
mu sync.RWMutex
}
type BroadcastMessage struct {
RoomID string
Message []byte
Exclude string
}
func NewHub() *Hub {
return &Hub{
Clients: make(map[string]*Client),
Rooms: make(map[string]map[string]*Client),
Register: make(chan *Client),
Unregister: make(chan *Client),
Broadcast: make(chan *BroadcastMessage, 256),
}
}
func (h *Hub) Run(ctx context.Context) {
for {
select {
case client := <-h.Register:
h.mu.Lock()
h.Clients[client.ID] = client
h.mu.Unlock()
log.Printf("Client connected: %s (user: %s)", client.ID, client.UserID)
case client := <-h.Unregister:
h.mu.Lock()
if _, ok := h.Clients[client.ID]; ok {
delete(h.Clients, client.ID)
close(client.Send)
// 从所有房间移除
client.mu.RLock()
for roomID := range client.Rooms {
if room, ok := h.Rooms[roomID]; ok {
delete(room, client.ID)
if len(room) == 0 {
delete(h.Rooms, roomID)
}
}
}
client.mu.RUnlock()
}
h.mu.Unlock()
log.Printf("Client disconnected: %s", client.ID)
case msg := <-h.Broadcast:
h.mu.RLock()
if room, ok := h.Rooms[msg.RoomID]; ok {
for clientID, client := range room {
if clientID == msg.Exclude {
continue
}
select {
case client.Send <- msg.Message:
default:
// 客户端发送队列满，断开连接
close(client.Send)
delete(room, clientID)
}
}
}
h.mu.RUnlock()
case <-ctx.Done():
return
}
}
}
func (h *Hub) JoinRoom(clientID, roomID string) {
h.mu.Lock()
defer h.mu.Unlock()
client, ok := h.Clients[clientID]
if !ok {
return
}
client.mu.Lock()
client.Rooms[roomID] = true
client.mu.Unlock()
if _, ok := h.Rooms[roomID]; !ok {
h.Rooms[roomID] = make(map[string]*Client)
}
h.Rooms[roomID][clientID] = client
}
func (h *Hub) LeaveRoom(clientID, roomID string) {
h.mu.Lock()
defer h.mu.Unlock()
client, ok := h.Clients[clientID]
if !ok {
return
}
client.mu.Lock()
delete(client.Rooms, roomID)
client.mu.Unlock()
if room, ok := h.Rooms[roomID]; ok {
delete(room, clientID)
if len(room) == 0 {
delete(h.Rooms, roomID)
}
}
}
var hub = NewHub()
var redisClient *redis.Client
func init() {
redisClient = redis.NewClient(&redis.Options{
Addr: "localhost:6379",
})
}
func handleWebSocket(w http.ResponseWriter, r *http.Request) {
// 验证JWT token
token := r.URL.Query().Get("token")
userID, err := validateToken(token)
if err != nil {
http.Error(w, "Unauthorized", http.StatusUnauthorized)
return
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Printf("Upgrade error: %v", err)
return
}
client := &Client{
ID: generateID(),
UserID: userID,
Conn: conn,
Send: make(chan []byte, 256),
Rooms: make(map[string]bool),
}
hub.Register <- client
// 读取消息
go client.readPump()
// 写入消息
go client.writePump()
}
func (c *Client) readPump() {
defer func() {
hub.Unregister <- c
c.Conn.Close()
}()
c.Conn.SetReadLimit(4096)
c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
c.Conn.SetPongHandler(func(string) error {
c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
return nil
})
for {
_, message, err := c.Conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
log.Printf("Read error: %v", err)
}
break
}
var msg Message
if err := json.Unmarshal(message, &msg); err != nil {
log.Printf("Unmarshal error: %v", err)
continue
}
msg.UserID = c.UserID
msg.Time = time.Now().UnixMilli()
switch msg.Type {
case "join":
hub.JoinRoom(c.ID, msg.RoomID)
c.Send <- createResponse("joined", msg.RoomID)
case "leave":
hub.LeaveRoom(c.ID, msg.RoomID)
c.Send <- createResponse("left", msg.RoomID)
case "message":
msgBytes, _ := json.Marshal(msg)
hub.Broadcast <- &BroadcastMessage{
RoomID: msg.RoomID,
Message: msgBytes,
Exclude: c.ID,
}
// 保存到数据库
go saveMessage(msg)
}
}
}
func (c *Client) writePump() {
ticker := time.NewTicker(30 * time.Second)
defer func() {
ticker.Stop()
c.Conn.Close()
}()
for {
select {
case message, ok := <-c.Send:
c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if !ok {
c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
return
}
w, err := c.Conn.NextWriter(websocket.TextMessage)
if err != nil {
return
}
w.Write(message)
if err := w.Close(); err != nil {
return
}
case <-ticker.C:
c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
return
}
}
}
}
func main() {
ctx := context.Background()
go hub.Run(ctx)
http.HandleFunc("/ws", handleWebSocket)
log.Println("Server starting on :8080")
log.Fatal(http.ListenAndServe(":8080", nil))
}
水平扩展方案
Redis Pub/Sub实现多服务器通信
// scaling/redis-adapter.js
const { createAdapter } = require('@socket.io/redis-adapter');
const { createClient } = require('redis');
class RedisAdapter {
constructor() {
this.pubClient = createClient({ url: process.env.REDIS_URL });
this.subClient = this.pubClient.duplicate();
this.pubClient.on('error', err => console.error('Redis Pub Error:', err));
this.subClient.on('error', err => console.error('Redis Sub Error:', err));
}
async init(io) {
await Promise.all([
this.pubClient.connect(),
this.subClient.connect()
]);
io.adapter(createAdapter(this.pubClient, this.subClient));
console.log('Redis adapter initialized');
}
// 跨服务器发送消息
async broadcastToRoom(room, event, data) {
await this.pubClient.publish(
`room:${room}`,
JSON.stringify({ event, data })
);
}
// 获取在线用户数（跨服务器）
async getOnlineUsers() {
const keys = await this.pubClient.keys('socket:user:*');
return keys.length;
}
// 获取房间成员（跨服务器）
async getRoomMembers(room) {
const members = await this.pubClient.sMembers(`room:${room}:members`);
return members;
}
}
module.exports = RedisAdapter;
// docker-compose.yml
version: '3.8'
services:
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redis-data:/data
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
depends_on:
- ws-server-1
- ws-server-2
- ws-server-3
ws-server-1:
build: .
environment:
- REDIS_URL=redis://redis:6379
- JWT_SECRET=${JWT_SECRET}
depends_on:
- redis
ws-server-2:
build: .
environment:
- REDIS_URL=redis://redis:6379
- JWT_SECRET=${JWT_SECRET}
depends_on:
- redis
ws-server-3:
build: .
environment:
- REDIS_URL=redis://redis:6379
- JWT_SECRET=${JWT_SECRET}
depends_on:
- redis
volumes:
redis-data:
Nginx负载均衡配置
# nginx.conf
upstream websocket_backend {
# 使用IP hash保证连接粘性
ip_hash;
server ws-server-1:3000 max_fails=3 fail_timeout=30s;
server ws-server-2:3000 max_fails=3 fail_timeout=30s;
server ws-server-3:3000 max_fails=3 fail_timeout=30s;
keepalive 32;
}
server {
listen 80;
server_name ws.example.com;
# WebSocket连接升级
location /ws {
proxy_pass http://websocket_backend;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# 超时配置
proxy_connect_timeout 7d;
proxy_send_timeout 7d;
proxy_read_timeout 7d;
# 缓冲配置
proxy_buffering off;
}
# 健康检查
location /health {
access_log off;
return 200 "healthy\n";
}
}
消息持久化与离线消息
// services/messageService.js
const { Pool } = require('pg');
const pool = new Pool({
connectionString: process.env.DATABASE_URL
});
class MessageService {
// 保存消息
async saveMessage(message) {
const query = `
INSERT INTO messages (room_id, user_id, content, type, timestamp)
VALUES ($1, $2, $3, $4, $5)
RETURNING id
`;
const result = await pool.query(query, [
message.roomId,
message.userId,
message.content,
message.type,
message.timestamp
]);
return result.rows[0].id;
}
// 获取历史消息
async getMessages(roomId, limit = 50, before = null) {
let query = `
SELECT m.*, u.username, u.avatar
FROM messages m
JOIN users u ON m.user_id = u.id
WHERE m.room_id = $1
`;
const params = [roomId];
if (before) {
query += ' AND m.timestamp < $2';
params.push(before);
}
query += ' ORDER BY m.timestamp DESC LIMIT $' + (params.length + 1);
params.push(limit);
const result = await pool.query(query, params);
return result.rows.reverse();
}
// 获取离线消息
async getOfflineMessages(userId, since) {
const query = `
SELECT m.*, r.name as room_name
FROM messages m
JOIN rooms r ON m.room_id = r.id
JOIN room_members rm ON r.id = rm.room_id
WHERE rm.user_id = $1
AND m.timestamp > $2
AND m.user_id != $1
ORDER BY m.timestamp ASC
`;
const result = await pool.query(query, [userId, since]);
return result.rows;
}
// 标记消息已读
async markAsRead(userId, roomId, messageId) {
const query = `
INSERT INTO message_reads (user_id, room_id, message_id, read_at)
VALUES ($1, $2, $3, NOW())
ON CONFLICT (user_id, message_id) DO NOTHING
`;
await pool.query(query, [userId, roomId, messageId]);
}
}
module.exports = new MessageService();
// server.js - 集成离线消息
io.on('connection', async (socket) => {
// 发送离线消息
const offlineMessages = await messageService.getOfflineMessages(
socket.userId,
socket.lastSeen || Date.now() - 7 * 24 * 60 * 60 * 1000 // 7天内
);
if (offlineMessages.length > 0) {
socket.emit('messages:offline', { messages: offlineMessages });
}
// 更新最后在线时间
socket.on('disconnect', async () => {
await updateUserLastSeen(socket.userId, Date.now());
});
});
性能优化与监控
// monitoring/metrics.js
const prometheus = require('prom-client');
// 定义指标
const connectedClients = new prometheus.Gauge({
name: 'websocket_connected_clients',
help: 'Number of connected WebSocket clients'
});
const messagesTotal = new prometheus.Counter({
name: 'websocket_messages_total',
help: 'Total number of WebSocket messages',
labelNames: ['type']
});
const messageDuration = new prometheus.Histogram({
name: 'websocket_message_duration_seconds',
help: 'Message processing duration',
labelNames: ['type'],
buckets: [0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1]
});
const roomMembers = new prometheus.Gauge({
name: 'websocket_room_members',
help: 'Number of members in each room',
labelNames: ['room']
});
// 更新指标
io.on('connection', (socket) => {
connectedClients.inc();
socket.on('message:send', (data) => {
const end = messageDuration.startTimer({ type: 'send' });
messagesTotal.inc({ type: 'send' });
// 处理消息...
end();
});
socket.on('join:room', (roomId) => {
roomMembers.inc({ room: roomId });
});
socket.on('leave:room', (roomId) => {
roomMembers.dec({ room: roomId });
});
socket.on('disconnect', () => {
connectedClients.dec();
});
});
// 暴露metrics端点
app.get('/metrics', async (req, res) => {
res.set('Content-Type', prometheus.register.contentType);
res.end(await prometheus.register.metrics());
});
// monitoring/health.js
app.get('/health', (req, res) => {
const health = {
status: 'ok',
timestamp: new Date().toISOString(),
uptime: process.uptime(),
connections: io.sockets.sockets.size,
memory: process.memoryUsage(),
cpu: process.cpuUsage()
};
res.json(health);
});
// 连接限制
const MAX_CONNECTIONS_PER_USER = 5;
io.use((socket, next) => {
const userConnections = Array.from(io.sockets.sockets.values())
.filter(s => s.userId === socket.userId)
.length;
if (userConnections >= MAX_CONNECTIONS_PER_USER) {
return next(new Error('Too many connections'));
}
next();
});
延伸阅读
- Socket.IO官方文档
- WebSocket协议RFC 6455
- Redis Pub/Sub
- Gorilla WebSocket
- High Performance Browser Networking - WebSockets
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。