Tokio Sender未实现Sink trait：如何无需手动循环转发通道？

核心问题在于：Tokio的broadcast::Sender和mpsc::Sender并没有直接实现futures::Sink trait，所以无法直接调用Stream::forward方法。这类同步原语与Sink抽象的桥接功能，由tokio-util工具库提供，而非单独的tokio-sink库。

解决方案步骤： #

1. 先在Cargo.toml中添加tokio-util依赖：

[dependencies]
tokio = { version = "1.0", features = ["full"] }
tokio-stream = "0.1"
tokio-util = { version = "0.7", features = ["sync"] }

2. 使用tokio_util::sync::BroadcastSink包装broadcast::Sender，使其实现Sink trait，之后就能用forward完成无循环转发：

use tokio::sync::broadcast;
use tokio_stream::wrappers::BroadcastStream;
use tokio_util::sync::BroadcastSink;

#[tokio::main]
async fn main() {
    let (tx0, rx0) = broadcast::channel::<u32>(10);
    let (tx1, mut rx1) = broadcast::channel::<u32>(10);
    
    tokio::task::spawn(async move {
        // 将broadcast Sender转换为Sink
        let sink = BroadcastSink::new(tx1);
        // 执行流转发并处理错误
        if let Err(e) = BroadcastStream::new(rx0).forward(sink).await {
            eprintln!("转发出错: {}", e);
        }
    });
    
    // 发送数据并处理发送错误
    if let Err(e) = tx0.send(1) {
        eprintln!("发送数据失败: {}", e);
    }
    
    // 接收并打印结果
    match rx1.recv().await {
        Ok(val) => println!("收到数据: {}", val),
        Err(e) => eprintln!("接收失败: {}", e),
    }
}

针对mpsc通道的补充： #

如果是tokio::sync::mpsc::Sender，可以用tokio_util::sync::MpscSink做同样的转换，用法与上述示例完全一致。

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