JavaScript系列（35）-- WebSocket应用详解

JavaScript WebSocket应用详解 📡

今天，让我们深入了解JavaScript的WebSocket应用，这是实现实时双向通信的关键技术。

WebSocket基础概念 🌟

💡 小知识：WebSocket是一种在单个TCP连接上进行全双工通信的协议。它提供了在Web客户端和服务器之间建立持久连接的标准方法，使得双向数据传输成为可能。

基本WebSocket实现 📊

// 1. 基础WebSocket客户端
class WebSocketClient {
    constructor(url, options = {}) {
        this.url = url;
        this.options = {
            reconnectAttempts: 5,
            reconnectDelay: 3000,
            heartbeatInterval: 30000,
            ...options
        };
        
        this.ws = null;
        this.reconnectCount = 0;
        this.handlers = new Map();
        
        this.connect();
    }
    
    connect() {
        try {
            this.ws = new WebSocket(this.url);
            this.setupEventListeners();
            this.startHeartbeat();
        } catch (error) {
            console.error('WebSocket connection failed:', error);
            this.handleReconnect();
        }
    }
    
    setupEventListeners() {
        this.ws.onopen = () => {
            console.log('WebSocket connected');
            this.reconnectCount = 0;
            this.emit('connected');
        };
        
        this.ws.onclose = () => {
            console.log('WebSocket closed');
            this.handleReconnect();
            this.emit('disconnected');
        };
        
        this.ws.onerror = (error) => {
            console.error('WebSocket error:', error);
            this.emit('error', error);
        };
        
        this.ws.onmessage = (event) => {
            try {
                const message = JSON.parse(event.data);
                this.handleMessage(message);
            } catch (error) {
                console.error('Message parsing error:', error);
            }
        };
    }
    
    handleReconnect() {
        if (this.reconnectCount < this.options.reconnectAttempts) {
            this.reconnectCount++;
            setTimeout(() => {
                console.log(`Reconnecting... Attempt ${this.reconnectCount}`);
                this.connect();
            }, this.options.reconnectDelay);
        }
    }
    
    startHeartbeat() {
        this.heartbeatInterval = setInterval(() => {
            if (this.ws.readyState === WebSocket.OPEN) {
                this.send('heartbeat', { timestamp: Date.now() });
            }
        }, this.options.heartbeatInterval);
    }
    
    on(event, handler) {
        if (!this.handlers.has(event)) {
            this.handlers.set(event, new Set());
        }
        this.handlers.get(event).add(handler);
    }
    
    off(event, handler) {
        if (this.handlers.has(event)) {
            this.handlers.get(event).delete(handler);
        }
    }
    
    emit(event, data) {
        if (this.handlers.has(event)) {
            this.handlers.get(event).forEach(handler => handler(data));
        }
    }
    
    send(type, data) {
        if (this.ws.readyState === WebSocket.OPEN) {
            this.ws.send(JSON.stringify({ type, data }));
        }
    }
    
    close() {
        clearInterval(this.heartbeatInterval);
        if (this.ws) {
            this.ws.close();
        }
    }
}

// 2. 消息处理器
class MessageHandler {
    constructor() {
        this.handlers = new Map();
    }
    
    register(type, handler) {
        this.handlers.set(type, handler);
    }
    
    async handle(message) {
        const handler = this.handlers.get(message.type);
        if (handler) {
            try {
                return await handler(message.data);
            } catch (error) {
                console.error(`Error handling message type ${message.type}:`, error);
                throw error;
            }
        }
        throw new Error(`Unknown message type: ${message.type}`);
    }
}

// 3. 重试机制
class RetryStrategy {
    constructor(maxAttempts = 5, baseDelay = 1000) {
        this.maxAttempts = maxAttempts;
        this.baseDelay = baseDelay;
    }
    
    getDelay(attempt) {
        // 指数退避策略
        return Math.min(
            this.baseDelay * Math.pow(2, attempt),
            30000 // 最大延迟30秒
        );
    }
    
    async retry(operation, context) {
        let lastError;
        
        for (let attempt = 0; attempt < this.maxAttempts; attempt++) {
            try {
                return await operation(context);
            } catch (error) {
                lastError = error;
                const delay = this.getDelay(attempt);
                console.log(`Retry attempt ${attempt + 1} after ${delay}ms`);
                await new Promise(resolve => setTimeout(resolve, delay));
            }
        }
        
        throw new Error(
            `Operation failed after ${this.maxAttempts} attempts: ${lastError}`
        );
    }
}

高级WebSocket应用 🚀

// 1. 实时数据流处理
class RealTimeDataStream {
    constructor(wsClient) {
        this.wsClient = wsClient;
        this.subscribers = new Map();
        this.buffer = [];
        this.bufferSize = 1000;
        
        this.setupHandlers();
    }
    
    setupHandlers() {
        this.wsClient.on('message', (data) => {
            this.handleData(data);
        });
    }
    
    handleData(data) {
        // 缓存数据
        this.buffer.push({
            timestamp: Date.now(),
            data
        });
        
        // 维护缓冲区大小
        if (this.buffer.length > this.bufferSize) {
            this.buffer.shift();
        }
        
        // 通知订阅者
        this.notifySubscribers(data);
    }
    
    subscribe(id, handler, filter = null) {
        this.subscribers.set(id, {
            handler,
            filter
        });
        
        // 发送缓冲区数据
        this.buffer
            .filter(item => !filter || filter(item.data))
            .forEach(item => handler(item.data));
    }
    
    unsubscribe(id) {
        this.subscribers.delete(id);
    }
    
    notifySubscribers(data) {
        for (const [id, { handler, filter }] of this.subscribers) {
            if (!filter || filter(data)) {
                handler(data);
            }
        }
    }
}

// 2. 消息队列实现
class MessageQueue {
    constructor() {
        this.queue = [];
        this.processing = false;
        this.maxRetries = 3;
    }
    
    async enqueue(message, priority = 0) {
        this.queue.push({
            message,
            priority,
            retries: 0
        });
        
        this.queue.sort((a, b) => b.priority - a.priority);
        
        if (!this.processing) {
            this.processQueue();
        }
    }
    
    async processQueue() {
        if (this.queue.length === 0) {
            this.processing = false;
            return;
        }
        
        this.processing = true;
        const item = this.queue.shift();
        
        try {
            await this.processMessage(item.message);
        } catch (error) {
            if (item.retries < this.maxRetries) {
                item.retries++;
                this.queue.push(item);
                console.log(`Message processing failed, retrying... (${item.retries}/${this.maxRetries})`);
            } else {
                console.error('Message processing failed after max retries:', error);
            }
        }
        
        this.processQueue();
    }
    
    async processMessage(message) {
        // 实现具体的消息处理逻辑
        return new Promise((resolve, reject) => {
            // 模拟消息处理
            setTimeout(() => {
                if (Math.random() > 0.1) { // 90%成功率
                    resolve();
                } else {
                    reject(new Error('Random processing error'));
                }
            }, 100);
        });
    }
}

// 3. 状态同步管理器
class StateSyncManager {
    constructor(wsClient) {
        this.wsClient = wsClient;
        this.state = {};
        this.version = 0;
        this.pendingUpdates = new Map();
        
        this.setupHandlers();
    }
    
    setupHandlers() {
        this.wsClient.on('state_update', (update) => {
            this.handleStateUpdate(update);
        });
        
        this.wsClient.on('state_sync', (fullState) => {
            this.handleStateSync(fullState);
        });
    }
    
    handleStateUpdate(update) {
        if (update.version > this.version) {
            this.applyUpdate(update);
            this.version = update.version;
            
            // 处理待处理的更新
            this.processPendingUpdates();
        } else {
            // 将更新添加到待处理队列
            this.pendingUpdates.set(update.version, update);
        }
    }
    
    handleStateSync(fullState) {
        this.state = fullState.state;
        this.version = fullState.version;
        this.pendingUpdates.clear();
    }
    
    applyUpdate(update) {
        // 递归合并更新
        this.state = this.mergeDeep(this.state, update.changes);
    }
    
    processPendingUpdates() {
        const sortedVersions = Array.from(this.pendingUpdates.keys()).sort();
        
        for (const version of sortedVersions) {
            if (version === this.version + 1) {
                const update = this.pendingUpdates.get(version);
                this.applyUpdate(update);
                this.version = version;
                this.pendingUpdates.delete(version);
            }
        }
    }
    
    mergeDeep(target, source) {
        const result = { ...target };
        
        for (const key in source) {
            if (typeof source[key] === 'object' && source[key] !== null) {
                result[key] = this.mergeDeep(result[key] || {}, source[key]);
            } else {
                result[key] = source[key];
            }
        }
        
        return result;
    }
}

性能优化技巧 ⚡

// 1. 消息批处理
class MessageBatcher {
    constructor(wsClient, options = {}) {
        this.wsClient = wsClient;
        this.options = {
            maxBatchSize: 100,
            maxDelay: 1000,
            ...options
        };
        
        this.batch = [];
        this.timer = null;
    }
    
    add(message) {
        this.batch.push(message);
        
        if (this.batch.length >= this.options.maxBatchSize) {
            this.flush();
        } else if (!this.timer) {
            this.timer = setTimeout(() => this.flush(), this.options.maxDelay);
        }
    }
    
    flush() {
        if (this.batch.length > 0) {
            this.wsClient.send('batch', this.batch);
            this.batch = [];
        }
        
        if (this.timer) {
            clearTimeout(this.timer);
            this.timer = null;
        }
    }
}

// 2. 消息压缩
class MessageCompressor {
    static compress(data) {
        // 使用简单的压缩策略
        return {
            t: data.type,
            d: data.data,
            ts: data.timestamp
        };
    }
    
    static decompress(compressed) {
        return {
            type: compressed.t,
            data: compressed.d,
            timestamp: compressed.ts
        };
    }
}

// 3. 连接池管理
class WebSocketPool {
    constructor(url, poolSize = 5) {
        this.url = url;
        this.poolSize = poolSize;
        this.pool = [];
        this.activeConnections = new Map();
        
        this.initialize();
    }
    
    async initialize() {
        for (let i = 0; i < this.poolSize; i++) {
            const ws = new WebSocketClient(this.url);
            this.pool.push(ws);
        }
    }
    
    acquire() {
        const ws = this.pool.pop();
        if (ws) {
            this.activeConnections.set(ws, Date.now());
            return ws;
        }
        throw new Error('No available connections in pool');
    }
    
    release(ws) {
        this.activeConnections.delete(ws);
        this.pool.push(ws);
    }
    
    cleanup() {
        const now = Date.now();
        for (const [ws, timestamp] of this.activeConnections) {
            if (now - timestamp > 30000) { // 30秒超时
                ws.close();
                this.activeConnections.delete(ws);
                const newWs = new WebSocketClient(this.url);
                this.pool.push(newWs);
            }
        }
    }
}

最佳实践建议 💡

1. 错误处理和恢复

// 1. 错误处理策略
class WebSocketErrorHandler {
    constructor(wsClient) {
        this.wsClient = wsClient;
        this.setupErrorHandlers();
    }
    
    setupErrorHandlers() {
        this.wsClient.on('error', (error) => {
            this.handleError(error);
        });
        
        window.addEventListener('online', () => {
            this.handleNetworkRecovery();
        });
    }
    
    handleError(error) {
        if (error instanceof WebSocket.CloseEvent) {
            this.handleConnectionError(error);
        } else if (error instanceof SyntaxError) {
            this.handleMessageError(error);
        } else {
            this.handleGenericError(error);
        }
    }
    
    handleConnectionError(error) {
        console.error('WebSocket connection error:', error);
        // 实现重连逻辑
    }
    
    handleMessageError(error) {
        console.error('Message parsing error:', error);
        // 实现消息重发逻辑
    }
    
    handleGenericError(error) {
        console.error('Generic WebSocket error:', error);
        // 实现通用错误处理逻辑
    }
    
    handleNetworkRecovery() {
        console.log('Network recovered, attempting to reconnect...');
        this.wsClient.connect();
    }
}

// 2. 安全性考虑
class WebSocketSecurity {
    constructor(wsClient) {
        this.wsClient = wsClient;
        this.setupSecurity();
    }
    
    setupSecurity() {
        // 添加消息验证
        this.wsClient.on('message', (message) => {
            if (!this.validateMessage(message)) {
                console.error('Invalid message received');
                return;
            }
            // 处理验证通过的消息
        });
    }
    
    validateMessage(message) {
        // 实现消息验证逻辑
        return message && 
               message.type &&
               typeof message.data !== 'undefined';
    }
    
    sanitizeData(data) {
        // 实现数据清理逻辑
        return JSON.parse(JSON.stringify(data)); // 深拷贝
    }
}

// 3. 性能监控
class WebSocketMonitor {
    constructor(wsClient) {
        this.wsClient = wsClient;
        this.metrics = {
            messagesSent: 0,
            messagesReceived: 0,
            errors: 0,
            latency: []
        };
        
        this.setupMonitoring();
    }
    
    setupMonitoring() {
        this.wsClient.on('message', () => {
            this.metrics.messagesReceived++;
        });
        
        this.wsClient.on('error', () => {
            this.metrics.errors++;
        });
    }
    
    measureLatency() {
        const start = Date.now();
        this.wsClient.send('ping', { timestamp: start });
        
        this.wsClient.on('pong', (data) => {
            const latency = Date.now() - data.timestamp;
            this.metrics.latency.push(latency);
        });
    }
    
    getMetrics() {
        return {
            ...this.metrics,
            averageLatency: this.calculateAverageLatency()
        };
    }
    
    calculateAverageLatency() {
        if (this.metrics.latency.length === 0) return 0;
        const sum = this.metrics.latency.reduce((a, b) => a + b, 0);
        return sum / this.metrics.latency.length;
    }
}

结语 📝

WebSocket技术为Web应用提供了强大的实时通信能力。通过本文，我们学习了：

1. WebSocket的基本概念和实现方法
2. 高级WebSocket应用场景
3. 性能优化技巧
4. 错误处理和安全性考虑
5. 最佳实践和监控策略

💡 学习建议：在使用WebSocket时，要特别注意连接的可靠性和消息的处理效率。合理使用重连机制、消息队列和批处理策略，可以显著提升应用的稳定性和性能。

如果你觉得这篇文章有帮助，欢迎点赞收藏，也期待在评论区看到你的想法和建议！👇

终身学习，共同成长。

咱们下一期见

💻