什么是JavaScript的异步生成器与WebSocket的结合，以及它如何实现实时数据流的异步迭代？

async function* createWebSocketMessageStream(url) {
    const ws = new WebSocket(url);
    let messageQueue = [];
    let resolveNextMessage = null; // 用于解决下一个Promise的函数

    // 当WebSocket接收到消息时
    ws.onmessage = (event) => {
        if (resolveNextMessage) {
            // 如果有等待的消费者，直接解决Promise
            resolveNextMessage(event.data);
            resolveNextMessage = null; // 重置，等待下一个消费者
        } else {
            // 否则，将消息加入队列
            messageQueue.push(event.data);
        }
    };

    // 错误处理
    ws.onerror = (error) => {
        console.error("WebSocket error:", error);
        // 可以在这里抛出错误，中断生成器
        // 或者尝试重连
    };

    // 连接关闭时
    ws.onclose = () => {
        console.log("WebSocket connection closed.");
        // 当连接关闭时，我们可能希望停止生成器
        // 这里可以考虑抛出一个特殊的错误或完成迭代
    };

    // 等待连接建立
    await new Promise((resolve, reject) => {
        ws.onopen = resolve;
        ws.onerror = reject; // 如果连接失败，也应该拒绝
    });
    console.log("WebSocket connected.");

    try {
        while (ws.readyState === WebSocket.OPEN) {
            if (messageQueue.length > 0) {
                // 如果队列中有消息，立即返回
                yield messageQueue.shift();
            } else {
                // 否则，创建一个Promise等待下一条消息
                const message = await new Promise(resolve => {
                    resolveNextMessage = resolve;
                });
                yield message;
            }
        }
    } finally {
        // 确保在生成器结束时关闭WebSocket
        if (ws.readyState === WebSocket.OPEN) {
            ws.close();
        }
        console.log("WebSocket stream closed.");
    }
}

// 如何使用这个流：
(async () => {
    try {
        const messageStream = createWebSocketMessageStream("ws://echo.websocket.org"); // 替换为你的WebSocket地址

        // 模拟发送一些数据
        const wsTest = new WebSocket("ws://echo.websocket.org");
        wsTest.onopen = () => {
            console.log("Test WS connected, sending messages...");
            wsTest.send("Hello from async generator test 1");
            setTimeout(() => wsTest.send("Hello from async generator test 2"), 1000);
            setTimeout(() => wsTest.close(), 2500); // 模拟关闭
        };

        for await (const message of messageStream) {
            console.log("Received via async generator:", message);
            if (message.includes("2")) {
                console.log("Received second message, breaking loop.");
                break; // 消费者可以随时停止消费
            }
        }
        console.log("Async generator loop finished.");
    } catch (error) {
        console.error("Error in main consumer:", error);
    }
})();

const ws = new WebSocket("ws://your.server");
ws.onmessage = (event) => {
    // 处理消息1
    parseMessage(event.data);
    // 可能还需要处理消息2，消息3...
    // 如果这些处理是异步的，很快就会嵌套很多回调
    someAsyncOperation(event.data, (result) => {
        anotherAsyncOperation(result, (finalData) => {
            // ... 噩梦开始
        });
    });
};

async function* createRobustWebSocketStream(url) {
    let ws;
    let messageQueue = [];
    let resolveNextMessage = null;
    let reconnectAttempts = 0;
    const MAX_RECONNECT_ATTEMPTS = 5;
    const RECONNECT_DELAY_BASE = 1000; // 1 second

    const connect = async () => {
        return new Promise((resolve, reject) => {
            ws = new WebSocket(url);

            ws.onopen = () => {
                console.log("WebSocket connected.");
                reconnectAttempts = 0; // 成功连接后重置尝试次数
                resolve();
            };

            ws.onmessage = (event) => {
                if (resolveNextMessage) {
                    resolveNextMessage(event.data);
                    resolveNextMessage = null;
                } else {
                    // TODO: 考虑队列最大长度，如果超过则丢弃或报错
                    messageQueue.push(event.data);
                }
            };

            ws.onerror = (error) => {
                console.error("WebSocket error:", error);
                // 拒绝连接Promise，或在已连接时抛出错误
                if (ws.readyState === WebSocket.CONNECTING) {
                    reject(error);
                }
            };

            ws.onclose = (event) => {
                console.log("WebSocket connection closed:", event.code, event.reason);
                // 触发重连逻辑，但不要立即结束生成器
            };
        });
    };

    while (true) { // 外部循环用于重连
        try {
            await connect(); // 尝试连接

            // 连接成功后，开始迭代消息
            while (ws.readyState === WebSocket.OPEN) {
                if (messageQueue.length > 0) {
                    yield messageQueue.shift();
                } else {
                    const message = await new Promise(resolve => {
                        resolveNextMessage = resolve;
                    });
                    yield message;
                }
            }
        } catch (error) {
            console.error("Connection attempt failed:", error);
        }

        // 如果连接中断，尝试重连
        if (reconnectAttempts < MAX_RECONNECT_ATTEMPTS) {
            reconnectAttempts++;
            const delay = RECONNECT_DELAY_BASE * Math.pow(2, reconnectAttempts - 1);
            console.log(`Attempting to reconnect in ${delay / 1000} seconds... (Attempt ${reconnectAttempts})`);
            await new Promise(resolve => setTimeout(resolve, delay));
        } else {
            console.error("Max reconnect attempts reached. Giving up.");
            throw new Error("WebSocket connection failed permanently."); // 最终抛出错误，结束生成器
        }
    }
}