JavaScript高级异步编程：使用Async Iterators实现实时数据流处理

2.1 异步可迭代协议

一个对象要成为异步可迭代对象，必须实现[Symbol.asyncIterator]方法，该方法返回一个异步迭代器对象。

const asyncIterable = {
    [Symbol.asyncIterator]() {
        let count = 0;
        return {
            async next() {
                await delay(100); // 模拟异步操作
                if (count < 5) {
                    return { value: `数据${count++}`, done: false };
                }
                return { done: true };
            }
        };
    }
};

2.2 for-await-of循环

专门为异步迭代器设计的循环语法，能够自动等待每个异步值。

async function processStream() {
    for await (const value of asyncIterable) {
        console.log('接收到:', value);
        // 可以在这里进行异步处理
        await processValue(value);
    }
    console.log('数据流结束');
}

3.1 场景描述

我们需要构建一个实时股票数据处理器，能够：

• 连接WebSocket获取实时股票数据
• 对数据进行实时过滤和转换
• 批量处理并发送到分析引擎
• 优雅处理连接中断和重连

3.2 实现异步数据流生成器

class StockDataStream {
    constructor(symbols, batchSize = 10) {
        this.symbols = symbols;
        this.batchSize = batchSize;
        this.ws = null;
        this.isConnected = false;
    }

    async *[Symbol.asyncIterator]() {
        await this.connect();
        
        let buffer = [];
        
        try {
            while (this.isConnected) {
                const data = await this.receiveData();
                
                if (data.type === 'trade') {
                    buffer.push(this.transformData(data));
                    
                    if (buffer.length >= this.batchSize) {
                        yield buffer;
                        buffer = [];
                    }
                }
                
                if (data.type === 'heartbeat') {
                    yield { type: 'heartbeat', timestamp: Date.now() };
                }
            }
        } finally {
            await this.cleanup();
        }
    }

    async connect() {
        return new Promise((resolve, reject) => {
            this.ws = new WebSocket('wss://api.stockdata.example.com');
            
            this.ws.onopen = () => {
                this.isConnected = true;
                this.ws.send(JSON.stringify({
                    action: 'subscribe',
                    symbols: this.symbols
                }));
                resolve();
            };
            
            this.ws.onmessage = (event) => {
                this.pendingData = JSON.parse(event.data);
                if (this.dataResolver) {
                    this.dataResolver(this.pendingData);
                    this.dataResolver = null;
                }
            };
        });
    }

    async receiveData() {
        return new Promise((resolve) => {
            if (this.pendingData) {
                const data = this.pendingData;
                this.pendingData = null;
                resolve(data);
            } else {
                this.dataResolver = resolve;
            }
        });
    }

    transformData(rawData) {
        return {
            symbol: rawData.s,
            price: parseFloat(rawData.p),
            volume: parseInt(rawData.v),
            timestamp: new Date(rawData.t),
            exchange: rawData.x
        };
    }

    async cleanup() {
        if (this.ws) {
            this.ws.close();
        }
        this.isConnected = false;
    }
}

3.3 实现数据处理管道

async function createDataPipeline() {
    // 创建过滤转换器
    async function* filterByVolume(stream, minVolume) {
        for await (const batch of stream) {
            const filtered = batch.filter(item => 
                item.volume >= minVolume
            );
            if (filtered.length > 0) {
                yield filtered;
            }
        }
    }

    // 创建聚合器
    async function* aggregateByMinute(stream) {
        let currentMinute = null;
        let minuteData = [];
        
        for await (const batch of stream) {
            for (const item of batch) {
                const itemMinute = Math.floor(item.timestamp.getTime() / 60000);
                
                if (currentMinute === null) {
                    currentMinute = itemMinute;
                }
                
                if (itemMinute === currentMinute) {
                    minuteData.push(item);
                } else {
                    // 输出上一分钟的数据
                    yield {
                        minute: new Date(currentMinute * 60000),
                        data: minuteData,
                        averagePrice: minuteData.reduce((sum, item) => 
                            sum + item.price, 0) / minuteData.length,
                        totalVolume: minuteData.reduce((sum, item) => 
                            sum + item.volume, 0)
                    };
                    
                    // 开始新的一分钟
                    currentMinute = itemMinute;
                    minuteData = [item];
                }
            }
        }
        
        // 输出最后一分钟的数据
        if (minuteData.length > 0) {
            yield {
                minute: new Date(currentMinute * 60000),
                data: minuteData,
                averagePrice: minuteData.reduce((sum, item) => 
                    sum + item.price, 0) / minuteData.length,
                totalVolume: minuteData.reduce((sum, item) => 
                    sum + item.volume, 0)
            };
        }
    }

    // 创建数据处理器
    async function processStockData(symbols) {
        const stream = new StockDataStream(symbols);
        const volumeFilter = filterByVolume(stream, 1000);
        const aggregator = aggregateByMinute(volumeFilter);
        
        try {
            for await (const minuteSummary of aggregator) {
                console.log(`[${minuteSummary.minute.toISOString()}]`);
                console.log(`平均价格: $${minuteSummary.averagePrice.toFixed(2)}`);
                console.log(`总交易量: ${minuteSummary.totalVolume}`);
                console.log(`交易笔数: ${minuteSummary.data.length}`);
                
                // 发送到分析服务
                await sendToAnalytics(minuteSummary);
                
                // 检查是否需要告警
                await checkAlerts(minuteSummary);
            }
        } catch (error) {
            console.error('数据处理管道错误:', error);
            await handlePipelineError(error);
        }
    }

    return { processStockData };
}

3.4 错误处理与重连机制

class ResilientStockStream extends StockDataStream {
    constructor(symbols, maxRetries = 3) {
        super(symbols);
        this.maxRetries = maxRetries;
        this.retryCount = 0;
    }

    async *[Symbol.asyncIterator]() {
        while (this.retryCount  
                                setTimeout(() => reject(new Error('接收超时')), 30000)
                            )
                        ]);
                        
                        if (data.type === 'trade') {
                            buffer.push(this.transformData(data));
                            
                            if (buffer.length >= this.batchSize) {
                                yield buffer;
                                buffer = [];
                            }
                        }
                        
                        if (data.type === 'heartbeat') {
                            yield { type: 'heartbeat', timestamp: Date.now() };
                        }
                    } catch (error) {
                        if (error.message === '接收超时') {
                            console.warn('数据接收超时，尝试重连...');
                            break; // 跳出内层循环，触发重连
                        }
                        throw error;
                    }
                }
            } catch (error) {
                console.error(`连接错误 (尝试 ${this.retryCount + 1}/${this.maxRetries}):`, error);
                
                this.retryCount++;
                
                if (this.retryCount > this.maxRetries) {
                    throw new Error(`超过最大重试次数: ${error.message}`);
                }
                
                // 指数退避重连
                const delayTime = Math.min(1000 * Math.pow(2, this.retryCount), 30000);
                console.log(`等待 ${delayTime}ms 后重连...`);
                await new Promise(resolve => setTimeout(resolve, delayTime));
                
                await this.cleanup();
            }
        }
    }
}

4.1 组合多个数据流

async function* mergeStreams(...streams) {
    const asyncIterators = streams.map(stream => 
        stream[Symbol.asyncIterator]()
    );
    
    const results = new Array(asyncIterators.length);
    let active = asyncIterators.length;
    
    // 初始化所有迭代器
    for (let i = 0; i  {
            results[index] = result;
            if (result.done) {
                active--;
            }
        });
    }
    
    while (active > 0) {
        // 等待任意一个迭代器有数据
        await Promise.race(
            results.map((result, index) => 
                result ? Promise.resolve(result) : new Promise(() => {})
            ).filter(Boolean)
        );
        
        // 找出有数据的迭代器
        for (let i = 0; i < results.length; i++) {
            const result = results[i];
            if (result && !result.done) {
                yield result.value;
                results[i] = null;
                pull(i);
            }
        }
    }
}

// 使用示例
async function monitorMultipleStocks() {
    const appleStream = new StockDataStream(['AAPL']);
    const googleStream = new StockDataStream(['GOOGL']);
    const mergedStream = mergeStreams(appleStream, googleStream);
    
    for await (const data of mergedStream) {
        console.log('合并数据:', data);
    }
}

4.2 背压控制（Backpressure）

class BackpressureController {
    constructor(maxQueueSize = 100) {
        this.maxQueueSize = maxQueueSize;
        this.queue = [];
        this.resolveNext = null;
        this.isPaused = false;
    }

    async push(item) {
        if (this.queue.length >= this.maxQueueSize) {
            this.isPaused = true;
            await new Promise(resolve => {
                this.resolveNext = resolve;
            });
        }
        
        this.queue.push(item);
        
        if (this.resolveNext) {
            this.resolveNext();
            this.resolveNext = null;
            this.isPaused = false;
        }
    }

    async *drain() {
        while (true) {
            if (this.queue.length > 0) {
                yield this.queue.shift();
            } else {
                await new Promise(resolve => {
                    const checkQueue = () => {
                        if (this.queue.length > 0) {
                            resolve();
                        } else {
                            setTimeout(checkQueue, 10);
                        }
                    };
                    checkQueue();
                });
            }
        }
    }
}

// 使用背压控制的数据流
async function* controlledStream(sourceStream) {
    const controller = new BackpressureController(50);
    
    // 生产者
    (async () => {
        for await (const item of sourceStream) {
            await controller.push(item);
        }
    })().catch(console.error);
    
    // 消费者
    yield* controller.drain();
}

5.1 内存管理

• 及时释放不再需要的引用
• 使用WeakMap存储临时数据
• 批量处理减少内存碎片

async function* memoryEfficientStream(stream) {
    // 使用块作用域确保变量及时回收
    for await (const batch of stream) {
        {
            // 处理数据
            const processed = processBatch(batch);
            yield processed;
            // processed离开作用域后可被GC回收
        }
        
        // 强制垃圾回收（仅用于调试）
        if (global.gc) {
            global.gc();
        }
    }
}

5.2 性能监控

class MonitoredAsyncIterable {
    constructor(iterable, name) {
        this.iterable = iterable;
        this.name = name;
        this.metrics = {
            startTime: null,
            itemsProcessed: 0,
            totalTime: 0
        };
    }

    async *[Symbol.asyncIterator]() {
        this.metrics.startTime = Date.now();
        
        for await (const item of this.iterable) {
            const start = Date.now();
            
            yield item;
            
            const duration = Date.now() - start;
            this.metrics.itemsProcessed++;
            this.metrics.totalTime += duration;
            
            // 每100个项目输出一次性能报告
            if (this.metrics.itemsProcessed % 100 === 0) {
                this.reportMetrics();
            }
        }
        
        this.reportMetrics();
    }

    reportMetrics() {
        const avgTime = this.metrics.totalTime / this.metrics.itemsProcessed;
        console.log(`[${this.name}] 性能报告:`);
        console.log(`处理项目数: ${this.metrics.itemsProcessed}`);
        console.log(`平均处理时间: ${avgTime.toFixed(2)}ms`);
        console.log(`总耗时: ${Date.now() - this.metrics.startTime}ms`);
    }
}

6.1 异步迭代器的优势

• 声明式编程：代码更清晰，更易维护
• 内存效率：按需生成数据，避免一次性加载
• 错误隔离：单个迭代器错误不会影响整个管道
• 组合性：易于构建复杂的数据处理管道

6.2 适用场景

• 实时数据流处理（股票、IoT、日志）
• 大文件分块读取
• 数据库查询结果流式处理
• API分页数据获取
• WebSocket消息处理

6.3 未来发展方向

随着JavaScript语言的发展，异步迭代器可能会与以下技术更深度集成：

• Web Streams API的进一步整合
• 与Observable模式的互操作
• 在Web Workers中的优化使用
• TypeScript更完善的类型支持