multithreading - Rust多线程异步Websocket服务器

Question

我想学习Rust，因此决定将其用于实际项目。

想法是拥有一台服务器

主线程A中的

产生一个新线程B，该线程执行一些异步任务，该任务通过时间

生成值流

异步接收客户端websocket连接[c, d, e, ..]并处理它们并生成新线程[C, D, E, ...]

将线程B中产生的值发送到线程[C, D, E, ...]

[C, D, E, ...]中的每个线程将值发布到[c, d, e, ..]中的相应客户端

我在用

tokio生成新线程，tokio::sync::mpsc::unbounded_channel将B中计算出的值发送给其他线程

tokio_tungstenite来管理websocket连接并将值发送到客户端

我设法得到一个工作示例，其中线程 B产生整数和固定的时间间隔。服务器启动时， B开始生成值 [0,1,2,3, ..]的流。

当打开一个新的Websocket连接时，客户端将从连接打开后产生的值开始接收数据流(这样，如果连接在 3生成值 B之后开始，则客户端将从以下位置接收值: 4及更高版本)。

这里是要捕获的地方。

我发现 C中 channel 的接收部分异步接收值的唯一方法(因此，防止在 c完全完成时才缓冲值并将它们发送到 B)是使用一个我认为消耗100的循环CPU的百分比。

我注意到，因此，每个websocket连接将消耗100％的CPU(因此，如果有两个连接，打开的CPU使用率将为200％，依此类推)。

这是循环:

loop {
    while let Ok(v) = rx.try_recv() {
       println!("PRINTER ID [{}] | RECEIVED: {:#?}", addr, v);
       println!("PRINTER ID [{}] | SENDING TO WS: {:#?}", addr, v);
       let mess = Message::Text(v.to_string());ws_sender.send(mess).await?;
}

如果我使用 recv()(而不是 try_recv())，则仅在完成 B时，这些值才会被缓冲并释放到websocket。

我尝试使用 futures_channel::unbounded而不是 tokio channel ，但是我有相同的缓冲区问题。

问题:如何重写以上循环，以避免在不阻塞的情况下将100％和流值用于websocket？

您可以在此处查看tokio服务器: https://github.com/ceikit/async_data/blob/master/src/bin/tokio_server.rs

您可以通过在另一个运行客户端的终端窗口中旋转Websocket连接来对其进行测试

Answer 1

需要将thread::sleep更改为使用futures-timer和sync::Mutex到futures::lock::Mutex，然后将带有while-let的recv()完美地工作