Mastering WebSocket: Full‑Duplex Communication, Server Push, and Real‑Time Messaging Implementations

1. WebSocket Overview

1.1 What is WebSocket

WebSocket is a protocol that provides full‑duplex communication over a single TCP connection, offering higher efficiency and real‑time interaction compared with HTTP polling.

1.2 WebSocket vs HTTP

Connection method: HTTP uses request‑response; WebSocket uses a persistent connection.

Direction: HTTP is half‑duplex; WebSocket is full‑duplex.

Server push: Not supported by HTTP; supported by WebSocket.

Overhead: HTTP adds headers on every request; WebSocket incurs overhead only during the initial handshake.

Real‑time: HTTP polling has high latency; WebSocket delivers real‑time data.

Typical scenarios: HTTP fits request‑response patterns; WebSocket fits real‑time interactive applications.

1.3 Application Scenarios

WebSocket 适用场景
├── 即时通讯        # 聊天消息、客服系统
├── 实时协作        # 文档协作、白板
├── 实时数据        # 股票行情、监控面板
├── 游戏            # 多人游戏、实时对战
├── 推送通知        # 系统通知、提醒
├── 在线状态        # 用户上下线、在线列表
└── IoT            # 设备状态、远程控制

2. WebSocket Protocol Details

2.1 Handshake Process

WebSocket 握手
┌─────────────────────────────────────────────────────────────┐
│  Client ──────────────────────────────────────────────► │
│  GET /ws HTTP/1.1                                         │
│  Host: server.example.com                                 │
│  Upgrade: websocket                                       │
│  Connection: Upgrade                                      │
│  Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==               │
│  Sec-WebSocket-Version: 13                                │
│                                                            │
│  Server ───────────────────────────────────────────────► │
│  HTTP/1.1 101 Switching Protocols                         │
│  Upgrade: websocket                                       │
│  Connection: Upgrade                                      │
│  Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=        │
│  握手完成后，进入 WebSocket 帧传输                         │
└─────────────────────────────────────────────────────────────┘

2.2 Frame Structure

WebSocket 帧格式
┌─────────────────────────────────────────────────────────────┐
│  ┌─────────┬─────────┬─────────┬─────────┬──────────────┐ │
│  │ Opcode  │ Flags   │ Mask    │ Payload │ Masking‑Key │ │
│  │ (4bit)  │ (4bit)  │ (1bit)  │ Length │            │ │
│  └─────────┴─────────┴─────────┴─────────┴──────────────┘ │
│  Opcode values: 0x0 continuation, 0x1 text, 0x2 binary, 0x8 close, 0x9 ping, 0xA pong
└─────────────────────────────────────────────────────────────┘

3. Server‑Side Implementations

3.1 Python (websockets library)

# 使用 websockets 库
import asyncio
import websockets
import json

async def echo(websocket, path):
    async for message in websocket:
        data = json.loads(message)
        await websocket.send(json.dumps({"type": "echo", "data": data}))

async def main():
    async with websockets.serve(echo, "localhost", 8765):
        print("WebSocket server started on ws://localhost:8765")
        await asyncio.Future()

asyncio.run(main())

3.2 FastAPI WebSocket

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import json

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            message = json.dumps({"client_id": client_id, "message": data})
            await manager.broadcast(message)
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(json.dumps({"client_id": client_id, "message": "离线"}))

3.3 Go (gorilla/websocket)

package main

import (
    "fmt"
    "log"
    "net/http"
    "github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{CheckOrigin: func(r *http.Request) bool {return true}}

func wsHandler(w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {log.Printf("upgrade error: %v", err); return}
    defer conn.Close()
    for {
        _, message, err := conn.ReadMessage()
        if err != nil {log.Printf("read error: %v", err); break}
        if err := conn.WriteMessage(websocket.TextMessage, message); err != nil {log.Printf("write error: %v", err); break}
    }
}

func main() {
    http.HandleFunc("/ws", wsHandler)
    fmt.Println("WebSocket server started on :8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

3.4 Java Spring WebSocket

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(myHandler(), "/ws").setAllowedOrigins("*");
    }
    @Bean
    public WebSocketHandler myHandler() {
        return new MyWebSocketHandler();
    }
}

public class MyWebSocketHandler extends TextWebSocketHandler {
    private final CopyOnWriteArrayList<WebSocketSession> sessions = new CopyOnWriteArrayList<>();
    @Override
    public void afterConnectionEstablished(WebSocketSession session) {sessions.add(session);}
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) {
        for (WebSocketSession s : sessions) {
            s.sendMessage(new TextMessage("Received: " + message.getPayload()));
        }
    }
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {sessions.remove(session);}
}

4. Real‑Time Messaging System

4.1 Chatroom Implementation (FastAPI)

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict, List
import json
from datetime import datetime

app = FastAPI()

class ChatRoom:
    def __init__(self):
        self.rooms: Dict[str, List[WebSocket]] = {}
    async def join_room(self, room_id: str, websocket: WebSocket):
        await websocket.accept()
        self.rooms.setdefault(room_id, []).append(websocket)
    def leave_room(self, room_id: str, websocket: WebSocket):
        if room_id in self.rooms:
            self.rooms[room_id].remove(websocket)
    async def broadcast(self, room_id: str, message: dict):
        if room_id in self.rooms:
            for connection in self.rooms[room_id]:
                await connection.send_json(message)

chat_room = ChatRoom()

@app.websocket("/ws/chat/{room_id}/{username}")
async def chat_websocket(websocket: WebSocket, room_id: str, username: str):
    await chat_room.join_room(room_id, websocket)
    await chat_room.broadcast(room_id, {"type": "system", "content": f"{username} 加入了聊天室", "timestamp": datetime.now().isoformat()})
    try:
        while True:
            data = await websocket.receive_text()
            msg = json.loads(data)
            await chat_room.broadcast(room_id, {"type": "message", "username": username, "content": msg.get("content"), "timestamp": datetime.now().isoformat()})
    except WebSocketDisconnect:
        chat_room.leave_room(room_id, websocket)
        await chat_room.broadcast(room_id, {"type": "system", "content": f"{username} 离开了聊天室", "timestamp": datetime.now().isoformat()})

4.2 Private Message Manager (Python)

class PrivateMessageManager:
    def __init__(self):
        self.user_connections: Dict[str, WebSocket] = {}
    def register(self, user_id: str, websocket: WebSocket):
        self.user_connections[user_id] = websocket
    def unregister(self, user_id: str):
        self.user_connections.pop(user_id, None)
    async def send_to_user(self, user_id: str, message: dict) -> bool:
        if user_id in self.user_connections:
            await self.user_connections[user_id].send_json(message)
            return True
        return False
    async def send_private(self, from_user: str, to_user: str, content: str):
        message = {"type": "private", "from": from_user, "content": content, "timestamp": datetime.now().isoformat()}
        sent = await self.send_to_user(to_user, message)
        if from_user in self.user_connections:
            await self.user_connections[from_user].send_json({**message, "sent": sent})
        return sent

4.3 Message Storage Design (SQL)

-- messages table
CREATE TABLE messages (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    room_id VARCHAR(64) NOT NULL,
    sender_id VARCHAR(64) NOT NULL,
    content TEXT NOT NULL,
    message_type ENUM('text','image','file','system') DEFAULT 'text',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_room_time (room_id, created_at),
    INDEX idx_sender (sender_id)
);

-- offline_messages table
CREATE TABLE offline_messages (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    receiver_id VARCHAR(64) NOT NULL,
    sender_id VARCHAR(64) NOT NULL,
    content TEXT NOT NULL,
    is_read BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_receiver_unread (receiver_id, is_read)
);

5. Online Status Management

5.1 Heartbeat Mechanism (Python)

class HeartbeatManager:
    def __init__(self, timeout: int = 30):
        self.timeout = timeout
        self.last_heartbeat: Dict[str, float] = {}
        self.user_status: Dict[str, str] = {}
    def record_heartbeat(self, user_id: str):
        import time
        self.last_heartbeat[user_id] = time.time()
        self.user_status[user_id] = "online"
    def is_online(self, user_id: str) -> bool:
        import time
        if user_id not in self.last_heartbeat:
            return False
        elapsed = time.time() - self.last_heartbeat[user_id]
        if elapsed > self.timeout:
            self.user_status[user_id] = "offline"
            return False
        return True
    def get_online_users(self) -> List[str]:
        import time
        now = time.time()
        online = []
        for uid, last in list(self.last_heartbeat.items()):
            if now - last <= self.timeout:
                online.append(uid)
            else:
                self.user_status[uid] = "offline"
        return online

5.2 Redis‑Based Presence Manager

import redis.asyncio as redis

class RedisPresenceManager:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379)
        self.online_key = "online_users"
        self.user_key = "user:{}:status"
    async def set_online(self, user_id: str):
        await self.redis.sadd(self.online_key, user_id)
        await self.redis.hset(self.user_key.format(user_id), mapping={"status": "online", "last_seen": datetime.now().isoformat()})
    async def set_offline(self, user_id: str):
        await self.redis.srem(self.online_key, user_id)
        await self.redis.hset(self.user_key.format(user_id), "status", "offline")
    async def get_online_users(self) -> List[str]:
        return await self.redis.smembers(self.online_key)
    async def is_online(self, user_id: str) -> bool:
        return await self.redis.sismember(self.online_key, user_id)