异步通道上的非阻塞接收？

Posted 2023-03-14

【中文标题】异步通道上的非阻塞接收？

【英文标题】：Non-blocking recv on an async channel?

【发布时间】：2020-06-20 00:05:30

【问题描述】：

我在futures::channel::mpsc::Receiver 中寻找与try_recv() 等效的内容，但找不到。有吗？有try_next()，但我不认为它符合我的要求。如果我在一个空的频道上多次调用它，我当然不希望它惊慌。

或者，是否可以在异步代码中使用 std::sync::mpsc::channel，只要我从不在通道上进行阻塞读取（即始终使用 try_recv()）？从异步函数发送给它并在同步线程中使用呢？

这里有一些关于我正在尝试做的事情的详细信息：

我正在构建一个玩家通过 websocket 连接到服务器的游戏。 websocket 端将是异步的，每个连接都会产生一个任务来循环传入消息并将它们放入通道中。该频道是我需要建议的内容 - 是否应该使用 futures::channel::mpsc::channel（并使游戏循环异步 - 但请参阅问题 1）或 std::sync::mpsc::channel（请参阅问题 2）。

游戏引擎将在单独的线程中运行，并且每帧检查一次该通道（每个玩家都有自己的通道）上的玩家输入。我想避免糟糕（或缓慢）的 websocket 客户端可能会淹没连接并减慢主游戏引擎的速度，因此我只想为每个玩家频道每帧处理一条传入消息。如果玩家自己的频道被填满，他们只会伤害自己。但是每帧检查一条传入消息需要是非阻塞的，因为它在游戏循环中运行。

那么根据上面的问题 2，以下是否可行？

异步 ​​websocket -> 同步通道 -> 同步游戏循环（反之亦然）

【问题讨论】：

futures::channel::mpsc::Receiver 是 stream/futures... 所以就这样做 rx.await :p

我想知道我是否可以这样做。有什么好的教程/示例/文档吗？我会做一些实验。谢谢！

不多rust-lang.github.io/async-book/01_getting_started/…

当我尝试做rx.await 时，我得到the trait bound futures_channel::mpsc::Receiver<PlayerMessageExternal>: core::future::future::Future is not satisfied。这令人困惑。是不是说 futures::channel::mpsc::Receiver 不暗示 Future ？还是futures v0.3和核心库不匹配？

我建议你使用 tokio docs.rs/tokio/0.2.13/tokio/sync/mpsc/fn.channel.html

【参考方案1】：

看起来try_recv() 尚未针对 async-std 实现，尽管它似乎并不遥远（目前正在审查中）。 https://github.com/async-rs/async-std/issues/579

正如@Stargateur 所述（请参阅原始问题下的 cmets），tokio 的 mpsc 实现支持try_recv()。

我已将我的代码切换为使用 tokio 0.2（以及用于 websockets 的 tokio_tungstenite），它运行良好。