如何从同步代码中读取tokio::sync::mpsc::Receiver的单个值？

解决方案思路 #

核心是把异步接收和同步消费解耦：用一个后台异步任务持续从Tokio的Receiver接收数据，再把数据转发到同步阻塞通道中，同步代码直接从这个同步通道里取数据即可，完全避开所有权转移的问题。

具体实现 #

这里用标准库的std::sync::mpsc作为中间同步通道，也可以换成crossbeam-channel（性能更优），代码逻辑一致：

use bytes::Bytes;
use std::sync::mpsc::{self, Receiver as SyncReceiver};
use tokio::runtime::Handle;
use tokio::sync::mpsc::Receiver as TokioReceiver;

pub struct MultiBuf {
    sync_recv: SyncReceiver<Bytes>,
    curr_chunk: Bytes,
}

pub fn new_multibuf(tokio_recv: TokioReceiver<Bytes>) -> MultiBuf {
    // 创建同步阻塞通道，用于异步→同步的数据中转
    let (sync_send, sync_recv) = mpsc::channel();

    // 获取当前Tokio Runtime的句柄，用于启动后台异步任务
    let handle = Handle::current();
    handle.spawn(async move {
        let mut tokio_recv = tokio_recv;
        // 持续从异步Receiver取数据，转发到同步通道
        while let Some(bytes) = tokio_recv.recv().await {
            // 如果同步通道的接收端已关闭（MultiBuf被销毁），就退出任务
            if sync_send.send(bytes).is_err() {
                break;
            }
        }
    });

    MultiBuf {
        sync_recv,
        curr_chunk: Bytes::new(),
    }
}

pub fn next_chunk(buf: &mut MultiBuf) -> &[u8] {
    match buf.sync_recv.recv() {
        Ok(v) => {
            buf.curr_chunk = v;
            &buf.curr_chunk[..]
        }
        Err(_) => &[], // 通道关闭，无更多数据
    }
}

方案优势 #

1. 解决所有权问题：Tokio的Receiver所有权完全交给后台异步任务，同步代码只持有同步通道的Receiver，无需克隆或转移原异步Receiver的所有权。
2. 匹配需求逻辑：next_chunk调用时才会阻塞等待新数据，完全符合“调用者准备好消费才接收”的要求。
3. 自动资源清理：当MultiBuf被销毁时，同步通道的Sender会被自动释放，后台任务发送数据时会收到错误并退出，不会泄露线程或任务。

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