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

Question

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

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

示例："；foo1"；以及"；bar1"；因此我们处理它们并形成"；foo1bar1"foo2"；收到("bar2"将在2sec内收到(，因此我们将其处理为"bar2；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非常有用。