websockets-realtime（实时网络套接字）

WebSockets & Real-Time

Comprehensive guide for building real-time applications.

Real-Time Technologies

Comparison

When to Use What

WEBSOCKETS:
✓ Chat applications
✓ Real-time collaboration
✓ Gaming
✓ Financial trading
✓ IoT dashboards
✓ Any bidirectional communication

SERVER-SENT EVENTS (SSE):
✓ Live feeds (news, sports)
✓ Notifications
✓ Progress updates
✓ Server-initiated updates only

LONG POLLING:
✓ Fallback when WebSocket unavailable
✓ Simple, infrequent updates
✓ Behind strict firewalls

WEBRTC:
✓ Video/audio calls
✓ Screen sharing
✓ Peer-to-peer file transfer

WebSocket Fundamentals

How WebSockets Work

HTTP Upgrade Handshake:
┌──────┐                      ┌──────┐
│Client│  GET /ws HTTP/1.1    │Server│
│      │  Upgrade: websocket  │      │
│      │ ──────────────────>  │      │
│      │                      │      │
│      │  HTTP/1.1 101        │      │
│      │  Switching Protocols │      │
│      │ <──────────────────  │      │
└──────┘                      └──────┘

After handshake:
┌──────┐                      ┌──────┐
│Client│ <═══════════════════>│Server│
│      │  Full-duplex TCP     │      │
│      │  Binary or text      │      │
└──────┘                      └──────┘

Client Implementation

// Basic WebSocket client
const ws = new WebSocket("wss://api.example.com/ws");

ws.onopen = () => {
  console.log("Connected");
  ws.send(JSON.stringify({ type: "subscribe", channel: "updates" }));
};

ws.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log("Received:", data);
};

ws.onerror = (error) => {
  console.error("WebSocket error:", error);
};

ws.onclose = (event) => {
  console.log("Disconnected:", event.code, event.reason);
};

// Send message
ws.send(JSON.stringify({ type: "message", content: "Hello!" }));

// Close connection
ws.close(1000, "Normal closure");

Reconnection Logic

class ReconnectingWebSocket {
  private ws: WebSocket | null = null;
  private reconnectAttempts = 0;
  private maxReconnectAttempts = 10;
  private reconnectDelay = 1000;

  constructor(private url: string) {
    this.connect();
  }

  private connect() {
    this.ws = new WebSocket(this.url);

    this.ws.onopen = () => {
      console.log("Connected");
      this.reconnectAttempts = 0;
    };

    this.ws.onclose = (event) => {
      if (event.code !== 1000) {
        this.reconnect();
      }
    };

    this.ws.onerror = () => {
      this.ws?.close();
    };
  }

  private reconnect() {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      console.error("Max reconnection attempts reached");
      return;
    }

    this.reconnectAttempts++;
    const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);

    console.log(
      `Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`,
    );

    setTimeout(() => this.connect(), delay);
  }

  send(data: unknown) {
    if (this.ws?.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(data));
    }
  }
}

Server Implementation (Node.js)

ws Library

import { WebSocketServer, WebSocket } from "ws";
import { createServer } from "http";

const server = createServer();
const wss = new WebSocketServer({ server });

// Track connected clients
const clients = new Set<WebSocket>();

wss.on("connection", (ws, request) => {
  console.log("Client connected");
  clients.add(ws);

  // Send welcome message
  ws.send(JSON.stringify({ type: "connected", clientCount: clients.size }));

  ws.on("message", (data) => {
    try {
      const message = JSON.parse(data.toString());
      handleMessage(ws, message);
    } catch (error) {
      ws.send(JSON.stringify({ type: "error", message: "Invalid JSON" }));
    }
  });

  ws.on("close", () => {
    clients.delete(ws);
    console.log("Client disconnected");
  });

  ws.on("error", (error) => {
    console.error("WebSocket error:", error);
  });

  // Heartbeat to detect stale connections
  ws.isAlive = true;
  ws.on("pong", () => {
    ws.isAlive = true;
  });
});

// Heartbeat interval
const heartbeatInterval = setInterval(() => {
  wss.clients.forEach((ws) => {
    if (!ws.isAlive) {
      return ws.terminate();
    }
    ws.isAlive = false;
    ws.ping();
  });
}, 30000);

wss.on("close", () => {
  clearInterval(heartbeatInterval);
});

function handleMessage(ws: WebSocket, message: any) {
  switch (message.type) {
    case "broadcast":
      broadcast(message.content);
      break;
    case "private":
      // Handle private messages
      break;
    default:
      ws.send(
        JSON.stringify({ type: "error", message: "Unknown message type" }),
      );
  }
}

function broadcast(content: any) {
  const message = JSON.stringify({ type: "broadcast", content });
  clients.forEach((client) => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(message);
    }
  });
}

server.listen(3000);

Socket.IO

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

const httpServer = createServer();
const io = new Server(httpServer, {
  cors: {
    origin: "https://example.com",
    methods: ["GET", "POST"],
  },
});

// Namespace for chat
const chat = io.of("/chat");

chat.on("connection", (socket) => {
  console.log("User connected:", socket.id);

  // Join room
  socket.on("join", (room: string) => {
    socket.join(room);
    socket.to(room).emit("user_joined", { userId: socket.id });
  });

  // Handle message
  socket.on("message", (data: { room: string; content: string }) => {
    chat.to(data.room).emit("message", {
      from: socket.id,
      content: data.content,
      timestamp: Date.now(),
    });
  });

  // Leave room
  socket.on("leave", (room: string) => {
    socket.leave(room);
    socket.to(room).emit("user_left", { userId: socket.id });
  });

  socket.on("disconnect", () => {
    console.log("User disconnected:", socket.id);
  });
});

// Authentication middleware
io.use((socket, next) => {
  const token = socket.handshake.auth.token;
  if (validateToken(token)) {
    socket.data.user = decodeToken(token);
    next();
  } else {
    next(new Error("Authentication error"));
  }
});

httpServer.listen(3000);

Server-Sent Events (SSE)

Server Implementation

import express from "express";

const app = express();

app.get("/events", (req, res) => {
  // Set SSE headers
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");

  // Send initial event
  res.write("event: connected\n");
  res.write('data: {"status": "connected"}\n\n');

  // Send periodic updates
  const interval = setInterval(() => {
    const data = JSON.stringify({
      timestamp: Date.now(),
      value: Math.random(),
    });
    res.write(`data: ${data}\n\n`);
  }, 1000);

  // Cleanup on disconnect
  req.on("close", () => {
    clearInterval(interval);
    res.end();
  });
});

app.listen(3000);

Client Implementation

const eventSource = new EventSource("/events");

eventSource.onopen = () => {
  console.log("SSE connection opened");
};

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log("Received:", data);
};

eventSource.addEventListener("connected", (event) => {
  console.log("Connected event:", event.data);
});

eventSource.onerror = (error) => {
  console.error("SSE error:", error);
  if (eventSource.readyState === EventSource.CLOSED) {
    // Reconnect logic if needed
  }
};

// Close connection
eventSource.close();

Message Protocols

JSON Message Format

// Define message types
interface BaseMessage {
  type: string;
  timestamp: number;
  id: string;
}

interface ChatMessage extends BaseMessage {
  type: "chat";
  room: string;
  content: string;
  sender: string;
}

interface PresenceMessage extends BaseMessage {
  type: "presence";
  status: "online" | "offline" | "away";
  userId: string;
}

interface ErrorMessage extends BaseMessage {
  type: "error";
  code: string;
  message: string;
}

type Message = ChatMessage | PresenceMessage | ErrorMessage;

// Type-safe message handling
function handleMessage(data: string) {
  const message: Message = JSON.parse(data);

  switch (message.type) {
    case "chat":
      displayChatMessage(message);
      break;
    case "presence":
      updateUserPresence(message);
      break;
    case "error":
      handleError(message);
      break;
  }
}

Binary Protocols

// For high-performance needs, use binary formats

// MessagePack
import { encode, decode } from "@msgpack/msgpack";

const encoded = encode({ type: "position", x: 100, y: 200 });
ws.send(encoded);

ws.onmessage = (event) => {
  const data = decode(event.data);
};

// Protocol Buffers
// Define schema in .proto file, generate types
// Smaller messages, faster serialization

Scaling WebSockets

Architecture

                    Load Balancer
                   (Sticky Sessions)
                         │
        ┌────────────────┼────────────────┐
        │                │                │
        ▼                ▼                ▼
   ┌─────────┐      ┌─────────┐      ┌─────────┐
   │ Server 1│      │ Server 2│      │ Server 3│
   │ (Node)  │      │ (Node)  │      │ (Node)  │
   └────┬────┘      └────┬────┘      └────┬────┘
        │                │                │
        └────────────────┼────────────────┘
                         │
                    ┌─────────┐
                    │  Redis  │
                    │ Pub/Sub │
                    └─────────┘

Redis Pub/Sub for Cross-Server Messages

import Redis from "ioredis";
import { WebSocketServer } from "ws";

const pub = new Redis();
const sub = new Redis();

const wss = new WebSocketServer({ port: 3000 });

// Subscribe to channel
sub.subscribe("broadcast");

// Forward Redis messages to local clients
sub.on("message", (channel, message) => {
  wss.clients.forEach((client) => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(message);
    }
  });
});

// Publish messages to Redis
function broadcast(message: object) {
  pub.publish("broadcast", JSON.stringify(message));
}

// Receive from WebSocket, publish to Redis
wss.on("connection", (ws) => {
  ws.on("message", (data) => {
    broadcast(JSON.parse(data.toString()));
  });
});

Socket.IO with Redis Adapter

import { Server } from "socket.io";
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();
io.adapter(createAdapter(pubClient, subClient));

// Now messages are automatically synchronized across servers
io.emit("notification", { message: "Hello all servers!" });

React Integration

Custom Hook

import { useEffect, useRef, useState, useCallback } from 'react';

interface UseWebSocketOptions {
  url: string;
  onMessage?: (data: any) => void;
  reconnect?: boolean;
}

export function useWebSocket(options: UseWebSocketOptions) {
  const { url, onMessage, reconnect = true } = options;
  const wsRef = useRef<WebSocket | null>(null);
  const [isConnected, setIsConnected] = useState(false);
  const [lastMessage, setLastMessage] = useState<any>(null);

  const connect = useCallback(() => {
    const ws = new WebSocket(url);

    ws.onopen = () => setIsConnected(true);

    ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      setLastMessage(data);
      onMessage?.(data);
    };

    ws.onclose = () => {
      setIsConnected(false);
      if (reconnect) {
        setTimeout(connect, 3000);
      }
    };

    wsRef.current = ws;
  }, [url, onMessage, reconnect]);

  useEffect(() => {
    connect();
    return () => {
      wsRef.current?.close();
    };
  }, [connect]);

  const send = useCallback((data: any) => {
    if (wsRef.current?.readyState === WebSocket.OPEN) {
      wsRef.current.send(JSON.stringify(data));
    }
  }, []);

  return { isConnected, lastMessage, send };
}

// Usage
function ChatComponent() {
  const { isConnected, lastMessage, send } = useWebSocket({
    url: 'wss://api.example.com/chat',
    onMessage: (data) => {
      console.log('New message:', data);
    },
  });

  return (
    <div>
      <span>Status: {isConnected ? 'Connected' : 'Disconnected'}</span>
      <button onClick={() => send({ type: 'message', content: 'Hello!' })}>
        Send
      </button>
    </div>
  );
}

Socket.IO Client

import { io, Socket } from 'socket.io-client';
import { createContext, useContext, useEffect, useState } from 'react';

const SocketContext = createContext<Socket | null>(null);

export function SocketProvider({ children }: { children: React.ReactNode }) {
  const [socket, setSocket] = useState<Socket | null>(null);

  useEffect(() => {
    const newSocket = io('https://api.example.com', {
      auth: { token: getAuthToken() },
    });

    setSocket(newSocket);

    return () => {
      newSocket.close();
    };
  }, []);

  return (
    <SocketContext.Provider value={socket}>{children}</SocketContext.Provider>
  );
}

export function useSocket() {
  const socket = useContext(SocketContext);
  if (!socket) {
    throw new Error('useSocket must be used within SocketProvider');
  }
  return socket;
}

// Usage
function ChatRoom({ roomId }: { roomId: string }) {
  const socket = useSocket();
  const [messages, setMessages] = useState<Message[]>([]);

  useEffect(() => {
    socket.emit('join', roomId);

    socket.on('message', (message: Message) => {
      setMessages((prev) => [...prev, message]);
    });

    return () => {
      socket.emit('leave', roomId);
      socket.off('message');
    };
  }, [socket, roomId]);

  const sendMessage = (content: string) => {
    socket.emit('message', { room: roomId, content });
  };

  return (/* render messages */);
}

Security

Authentication

// Token-based authentication
const ws = new WebSocket("wss://api.example.com/ws");

ws.onopen = () => {
  // Send auth message immediately
  ws.send(
    JSON.stringify({
      type: "auth",
      token: localStorage.getItem("token"),
    }),
  );
};

// Server-side validation
wss.on("connection", (ws, request) => {
  let authenticated = false;
  const authTimeout = setTimeout(() => {
    if (!authenticated) {
      ws.close(4001, "Authentication timeout");
    }
  }, 5000);

  ws.on("message", (data) => {
    const message = JSON.parse(data.toString());

    if (message.type === "auth") {
      if (validateToken(message.token)) {
        authenticated = true;
        clearTimeout(authTimeout);
        ws.send(JSON.stringify({ type: "auth_success" }));
      } else {
        ws.close(4002, "Invalid token");
      }
    } else if (!authenticated) {
      ws.close(4003, "Not authenticated");
    }
  });
});

Rate Limiting

const rateLimits = new Map<WebSocket, { count: number; timestamp: number }>();

function checkRateLimit(ws: WebSocket): boolean {
  const now = Date.now();
  const limit = rateLimits.get(ws);

  if (!limit || now - limit.timestamp > 1000) {
    rateLimits.set(ws, { count: 1, timestamp: now });
    return true;
  }

  if (limit.count >= 10) {
    // 10 messages per second
    return false;
  }

  limit.count++;
  return true;
}

ws.on("message", (data) => {
  if (!checkRateLimit(ws)) {
    ws.send(JSON.stringify({ type: "error", message: "Rate limit exceeded" }));
    return;
  }
  // Process message
});

Best Practices

DO:

• Use WSS (WebSocket Secure) in production
• Implement heartbeat/ping-pong
• Handle reconnection gracefully
• Authenticate connections
• Validate all incoming messages
• Use message IDs for acknowledgment
• Implement backpressure handling
• Monitor connection health

DON'T:

• Trust client data without validation
• Send sensitive data without encryption
• Keep connections open indefinitely
• Ignore disconnection handling
• Block the message handler
• Send unbounded data
• Forget about horizontal scaling

Troubleshooting

Common Issues

Debug Logging

// Development logging
if (process.env.NODE_ENV === "development") {
  ws.on("message", (data) => {
    console.log("← Received:", JSON.parse(data.toString()));
  });

  const originalSend = ws.send.bind(ws);
  ws.send = (data: string) => {
    console.log("→ Sending:", JSON.parse(data));
    originalSend(data);
  };
}