如何在单个任务中同时读取来自多个 Tokio 频道的消息？

Question

我想同时读取和处理来自两个通道的消息并构建另一个消息并通过另一个通道发送此消息。

来自两个频道的消息以不同的频率接收（根据 sleep）。

示例：接收到“foo1”和“bar1”，因此我们处理它们并形成“foo1bar1”。收到“foo2”（“bar2”将在 2 秒内收到），因此我们将其处理为“foo2bar1”。收到“foo3”，因此构造“foo3bar1”。当收到“bar2”时，我们得到“foo4bar2”等等。

在目前的实现中，由于这两个任务不相互通信，我无法进行“fooNbarM”构造。

use std::time::Duration;
use tokio;
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tokio::time::sleep;
use futures::future::join_all;

async fn message_sender(msg: &'static str, foo_tx: UnboundedSender<Result<&str, Box<dyn std::error::Error + Send>>>) {
    loop {
        match foo_tx.send(Ok(msg)) {
            Ok(()) => {
                if msg == "foo" {
                    sleep(Duration::from_millis(1000)).await;
                } else {
                    sleep(Duration::from_millis(3000)).await;
                }
            }
            Err(_) => {
                println!("failed to send foo");
                break;
            }
        }
    }
}

#[tokio::main]
async fn main() {
    let result: Vec<&str> = vec![];

    let (foo_tx, mut foo_rx): (
        UnboundedSender<Result<&str, Box<dyn std::error::Error + Send>>>,
        UnboundedReceiver<Result<&str, Box<dyn std::error::Error + Send>>>,
    ) = tokio::sync::mpsc::unbounded_channel();
    let (bar_tx, mut bar_rx): (
        UnboundedSender<Result<&str, Box<dyn std::error::Error + Send>>>,
        UnboundedReceiver<Result<&str, Box<dyn std::error::Error + Send>>>,
    ) = tokio::sync::mpsc::unbounded_channel();

    let foo_sender_handle = tokio::spawn(async move {
        message_sender("foo", foo_tx).await;
    });

    let foo_handle = tokio::spawn(async move {
        while let Some(v) = foo_rx.recv().await {
            println!("{:?}", v);
        }
    });

    let bar_sender_handle = tokio::spawn(async move {
        message_sender("bar", bar_tx).await;
    });

    let bar_handle = tokio::spawn(async move {
        while let Some(v) = bar_rx.recv().await {
            println!("{:?}", v);
        }
    });

    let handles = vec![foo_sender_handle, foo_handle, bar_sender_handle, bar_handle];
    join_all(handles.into_iter()).await;
}

Cargo.toml

[package]
name = "play"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
tokio = { version = "1.16.1", features = ["full"] }
futures = "0.3.21"

Answer 1

使用tokio::select等待任一通道准备就绪：

use futures::future; // 0.3.19
use std::time::Duration;
use tokio::{
    sync::mpsc::{self, UnboundedSender},
    time,
}; // 1.16.1

async fn message_sender(msg: &'static str, foo_tx: UnboundedSender<String>) {
    for count in 0.. {
        let message = format!("{msg}{count}");
        foo_tx.send(message).unwrap();

        if msg == "foo" {
            time::sleep(Duration::from_millis(100)).await;
        } else {
            time::sleep(Duration::from_millis(300)).await;
        }
    }
}

#[tokio::main]
async fn main() {
    let (foo_tx, mut foo_rx) = mpsc::unbounded_channel();
    let (bar_tx, mut bar_rx) = mpsc::unbounded_channel();

    let foo_sender_handle = tokio::spawn(message_sender("foo", foo_tx));
    let bar_sender_handle = tokio::spawn(message_sender("bar", bar_tx));

    let receive_handle = tokio::spawn(async move {
        let mut foo = None;
        let mut bar = None;

        loop {
            tokio::select! {
                f = foo_rx.recv() => foo = f,
                b = bar_rx.recv() => bar = b,
            }

            if let (Some(foo), Some(bar)) = (&foo, &bar) {
                println!("{foo}{bar}");
            }
        }
    });

    future::join_all([foo_sender_handle, bar_sender_handle, receive_handle]).await;
}

您还必须处理仅收到一条消息的情况，因此 Option 很有用。

如何在单个任务中同时读取来自多个 Tokio 频道的消息？

How do I simultaneously read messages from multiple Tokio channels in a single task?

rust

rust-tokio