JavaScript高级异步编程实战：构建高性能实时数据流处理系统 | 现代前端开发深度解析

系统核心组件

• 数据源管理器：处理多个WebSocket和API数据源
• 流处理器：实时数据转换和过滤
• 任务调度器：并发任务管理和优先级控制
• 缓存层：内存和IndexedDB双重缓存策略
• 监控系统：性能指标和错误追踪

1. 高级Promise模式：可取消的异步操作

class CancelablePromise {
    constructor(executor) {
        this._isCanceled = false;
        this._promise = new Promise((resolve, reject) => {
            executor(
                value => !this._isCanceled && resolve(value),
                reason => !this._isCanceled && reject(reason)
            );
        });
    }

    cancel(reason = 'Operation canceled') {
        this._isCanceled = true;
        return Promise.reject(new Error(reason));
    }

    then(onFulfilled, onRejected) {
        return this._promise.then(onFulfilled, onRejected);
    }

    catch(onRejected) {
        return this._promise.catch(onRejected);
    }

    static raceWithCancel(promises, timeout = 5000) {
        const cancelToken = { current: null };
        const timeoutPromise = new Promise((_, reject) => {
            cancelToken.current = setTimeout(() => {
                reject(new Error(`Timeout after ${timeout}ms`));
            }, timeout);
        });

        return Promise.race([...promises, timeoutPromise]).finally(() => {
            clearTimeout(cancelToken.current);
        });
    }
}

// 使用示例
const dataFetch = new CancelablePromise((resolve) => {
    const timer = setTimeout(() => resolve('Data loaded'), 3000);
    return () => clearTimeout(timer); // 清理函数
});

// 需要时取消
setTimeout(() => dataFetch.cancel(), 1000);

2. 数据流处理器：基于生成器的管道模式

class DataStreamProcessor {
    constructor() {
        this._transformers = [];
        this._buffer = [];
        this._isProcessing = false;
    }

    *createDataGenerator(source) {
        for await (const chunk of source) {
            yield chunk;
        }
    }

    addTransformer(transformer) {
        this._transformers.push(transformer);
        return this;
    }

    async *processStream(source) {
        const generator = this.createDataGenerator(source);
        
        for await (let data of generator) {
            // 应用所有转换器
            for (const transform of this._transformers) {
                data = await transform(data);
            }
            
            yield data;
            
            // 流量控制：防止内存溢出
            if (this._buffer.length > 1000) {
                await this._throttle();
            }
        }
    }

    async _throttle() {
        return new Promise(resolve => {
            setTimeout(resolve, 100);
        });
    }

    // 批量处理优化
    async processInBatches(source, batchSize = 100) {
        let batch = [];
        
        for await (const item of this.processStream(source)) {
            batch.push(item);
            
            if (batch.length >= batchSize) {
                await this._processBatch(batch);
                batch = [];
            }
        }
        
        if (batch.length > 0) {
            await this._processBatch(batch);
        }
    }
}

// 使用示例
const processor = new DataStreamProcessor()
    .addTransformer(data => data.filter(item => item.active))
    .addTransformer(data => data.map(item => ({ ...item, processed: true })));

3. Web Workers并行计算

class ParallelProcessor {
    constructor(workerCount = navigator.hardwareConcurrency || 4) {
        this.workers = [];
        this.taskQueue = [];
        this.initializeWorkers(workerCount);
    }

    initializeWorkers(count) {
        const workerScript = `
            self.onmessage = async function(e) {
                const { id, data, operation } = e.data;
                
                try {
                    // 模拟复杂计算
                    const result = await processData(data, operation);
                    self.postMessage({ id, result, status: 'success' });
                } catch (error) {
                    self.postMessage({ id, error: error.message, status: 'error' });
                }
            };

            async function processData(data, operation) {
                // 实际的数据处理逻辑
                switch(operation) {
                    case 'filter':
                        return data.filter(item => item.value > 50);
                    case 'transform':
                        return data.map(item => ({ ...item, score: item.value * 2 }));
                    case 'aggregate':
                        return data.reduce((sum, item) => sum + item.value, 0);
                    default:
                        return data;
                }
            }
        `;

        for (let i = 0; i  {
            const taskId = Date.now() + Math.random();
            const task = { id: taskId, data, operation, resolve, reject };
            
            this.taskQueue.push(task);
            this.processQueue();
        });
    }

    processQueue() {
        const availableWorker = this.workers.find(w => !w.busy);
        
        if (availableWorker && this.taskQueue.length > 0) {
            const task = this.taskQueue.shift();
            availableWorker.busy = true;
            
            availableWorker.worker.postMessage({
                id: task.id,
                data: task.data,
                operation: task.operation
            });
        }
    }

    handleWorkerMessage(e) {
        const { id, result, error, status } = e.data;
        const workerIndex = this.workers.findIndex(w => !w.busy);
        
        if (workerIndex !== -1) {
            this.workers[workerIndex].busy = false;
        }
        
        this.processQueue();
        
        // 找到对应的任务并处理结果
        // ... 结果处理逻辑
    }
}

系统架构图

核心实现代码

class StockMonitor {
    constructor() {
        this.dataStreams = new Map();
        this.processors = new ParallelProcessor();
        this.cache = new DataCache();
        this.metrics = new PerformanceMetrics();
        this.initialize();
    }

    async initialize() {
        // 连接多个数据源
        await this.connectDataSources();
        
        // 启动数据处理管道
        this.startProcessingPipeline();
        
        // 启动监控
        this.startMonitoring();
    }

    async connectDataSources() {
        const sources = [
            'wss://api.stock1.com/stream',
            'wss://api.stock2.com/live',
            'wss://api.stock3.com/realtime'
        ];

        const connections = sources.map((url, index) => 
            this.createWebSocketConnection(url, `source_${index}`)
        );

        // 使用Promise.any获取最快响应
        const firstResponse = await Promise.any(
            connections.map(conn => conn.ready)
        );
        
        console.log('First data source connected:', firstResponse);
    }

    createWebSocketConnection(url, id) {
        return new Promise((resolve) => {
            const ws = new WebSocket(url);
            const connection = {
                ws,
                ready: new Promise(r => ws.onopen = () => r(id)),
                buffer: [],
                isActive: true
            };

            ws.onmessage = (event) => {
                const data = JSON.parse(event.data);
                this.handleIncomingData(id, data);
            };

            ws.onerror = (error) => {
                console.error(`WebSocket error for ${id}:`, error);
                this.handleConnectionLoss(id);
            };

            this.dataStreams.set(id, connection);
            resolve(connection);
        });
    }

    async handleIncomingData(sourceId, rawData) {
        // 性能标记开始
        this.metrics.mark(`process_${sourceId}_start`);
        
        try {
            // 并行处理数据
            const processedData = await this.processors.executeTask(
                rawData, 
                'transform'
            );
            
            // 缓存结果
            await this.cache.set(`stock_${Date.now()}`, processedData);
            
            // 触发更新事件
            this.dispatchUpdateEvent(processedData);
            
            // 记录性能指标
            this.metrics.measure(`process_${sourceId}`, {
                dataSize: JSON.stringify(rawData).length,
                processingTime: Date.now()
            });
            
        } catch (error) {
            this.handleProcessingError(sourceId, error);
        }
    }

    startProcessingPipeline() {
        // 创建数据处理管道
        const pipeline = new DataStreamProcessor()
            .addTransformer(this.validateData.bind(this))
            .addTransformer(this.enrichData.bind(this))
            .addTransformer(this.calculateMetrics.bind(this));
        
        // 启动管道处理
        this.processingPipeline = pipeline;
    }

    async validateData(data) {
        // 数据验证逻辑
        if (!data || !Array.isArray(data)) {
            throw new Error('Invalid data format');
        }
        
        return data.filter(item => 
            item.symbol && 
            typeof item.price === 'number' &&
            item.timestamp
        );
    }

    async enrichData(data) {
        // 数据增强：添加技术指标
        return data.map(item => ({
            ...item,
            movingAverage: this.calculateMA(item.history),
            rsi: this.calculateRSI(item.history),
            timestamp: new Date().toISOString()
        }));
    }

    // 更多数据处理方法...
}

// 性能监控类
class PerformanceMetrics {
    constructor() {
        this.metrics = new Map();
        this.reportInterval = setInterval(() => {
            this.reportMetrics();
        }, 60000); // 每分钟报告一次
    }

    mark(name) {
        performance.mark(`${name}_start`);
    }

    measure(name, metadata = {}) {
        performance.mark(`${name}_end`);
        performance.measure(name, `${name}_start`, `${name}_end`);
        
        const measures = performance.getEntriesByName(name);
        const lastMeasure = measures[measures.length - 1];
        
        this.metrics.set(name, {
            duration: lastMeasure.duration,
            timestamp: Date.now(),
            ...metadata
        });
    }

    reportMetrics() {
        const report = {
            timestamp: new Date().toISOString(),
            metrics: Array.from(this.metrics.entries())
        };
        
        // 发送到监控服务
        this.sendToMonitoringService(report);
        
        // 清理旧数据
        this.cleanupOldMetrics();
    }
}

单元测试示例

// 使用Jest进行测试
describe('DataStreamProcessor', () => {
    let processor;
    
    beforeEach(() => {
        processor = new DataStreamProcessor();
    });
    
    test('should process data through transformers', async () => {
        const mockSource = (async function*() {
            yield { value: 10, active: true };
            yield { value: 20, active: false };
        })();
        
        processor.addTransformer(data => ({ ...data, processed: true }));
        
        const results = [];
        for await (const data of processor.processStream(mockSource)) {
            results.push(data);
        }
        
        expect(results).toHaveLength(1);
        expect(results[0].processed).toBe(true);
    });
    
    test('should handle backpressure', async () => {
        // 测试流量控制
        const largeSource = createLargeDataSource(10000);
        await expect(processor.processInBatches(largeSource))
            .resolves.not.toThrow();
    });
});

推荐学习路径

• 深入理解Event Loop和Microtask Queue
• 学习RxJS响应式编程
• 探索Service Workers和PWA
• 研究WebAssembly性能优化