使用Boost Asio处理多WebSocket并发读取的问题及改进方案咨询

需要同时从多个WebSocket读取数据，需求如下：

• 各Socket相互独立
• 不同Socket的输入频率差异极大
• 偶尔需要根据特定输入将某一Socket的数据转发至另一Socket

我采用的方案是使用awaitable的operator ||并行等待读取操作，伪代码如下：

#include <boost/asio/experimental/awaitable_operators.hpp>

using namespace boost::asio::experimental::awaitable_operators;

auto stream1 = provider->get_stream_type1();
auto stream2 = provider->get_stream_type2();

while (true) {
    auto parallel_group = co_await (stream1->read() || stream2->read());

    const auto completed_index = parallel_group.index();

    std::string response{};

    if (completed_index == 0) response = std::get<0>(parallel_group);
    else if (completed_index == 1) response = std::get<1>(parallel_group);
    else throw std::runtime_error("unknown index");

    process(response);
}

但存在一个问题：operator ||会在任一读取操作成功时取消所有未完成的操作，这会导致我需要重新连接WebSocket，而重连开销极高，显然不合理。请问我哪里做得不对？有什么更好的实现方案？

你的问题核心在于误用了operator ||的语义——这个操作符的设计目标就是当任意一个awaitable完成时，取消其他所有未完成的操作，完全不符合你需要保持多个WebSocket连接持续活跃的需求。

正确的思路是给每个WebSocket单独分配独立协程，让它们各自持续处理读取操作，相互之间互不干扰；同时通过线程安全的消息队列或通道，实现不同Socket间的数据转发逻辑。

具体实现步骤 #

1. 为每个WebSocket创建独立协程
   每个协程单独维护对应WebSocket的连接，持续执行读取、数据处理逻辑，不依赖其他Socket的状态。示例代码框架：

   boost::asio::awaitable<void> handle_websocket(std::shared_ptr<WebSocketStream> ws, 
                                                 std::shared_ptr<ThreadSafeQueue<std::string>> forward_queue,
                                                 int stream_id) {
       try {
           while (true) {
               std::string response = co_await ws->read();
               // 处理当前Socket的收到的数据
               process(response);
               // 判断是否需要转发，若需要则放入目标队列
               if (need_forward(response, stream_id)) {
                   forward_queue->push(response);
               }
           }
       } catch (const std::exception& e) {
           // 单独处理当前Socket的连接异常，按需实现重连逻辑
           handle_reconnect(ws, stream_id);
       }
   }
   

2. 实现数据转发逻辑
   如果需要将A Socket的数据转发到B Socket，可在A的处理协程中把待转发数据放入线程安全队列；同时在B的协程中，除了处理自身WebSocket的读取操作，还额外监听转发队列的消息（可以用co_await (ws->read() || queue->pop())，注意pop需实现为可取消的awaitable），收到队列消息后直接通过B Socket发送。

3. 启动所有协程并行运行
   在主逻辑中，为每个WebSocket启动对应的处理协程，让它们独立执行：

   boost::asio::awaitable<void> main() {
       auto stream1 = provider->get_stream_type1();
       auto stream2 = provider->get_stream_type2();
       auto forward_queue = std::make_shared<ThreadSafeQueue<std::string>>();
   
       // 启动两个独立协程， detached 表示无需等待协程完成
       co_spawn(co_await boost::asio::this_coro::executor, 
                handle_websocket(stream1, forward_queue, 1), 
                boost::asio::detached);
       co_spawn(co_await boost::asio::this_coro::executor, 
                handle_websocket(stream2, forward_queue, 2), 
                boost::asio::detached);
   
       // 保持主协程持续运行，避免程序退出
       co_await boost::asio::steady_timer(co_await boost::asio::this_coro::executor, 
                                          std::chrono::hours(24)).async_wait(boost::asio::use_awaitable);
   }
   

额外注意事项 #

• 线程安全：多个协程共享的资源（如转发队列）必须保证线程安全，可通过Boost.Asio的strand或标准库的互斥锁+条件变量实现线程安全队列。
• 异常隔离：每个WebSocket协程单独处理自身的连接异常，避免单个Socket崩溃影响其他Socket的运行。
• 优雅取消：若需要关闭某个Socket，可给对应协程传入取消令牌，触发优雅退出，避免强制中断导致资源泄漏。

内容的提问来源于stack exchange，提问作者CaptainTrunky