如何動態(tài)匹配 Tokio 派生線程？你知道嗎？
當(dāng)我們在使用tokio和MPSC(多生產(chǎn)者單消費(fèi)者)通道時，通常以某種固定的方式連接派生線程。然而，在最近項(xiàng)目中，必須在各種配置中動態(tài)匹配異步生產(chǎn)者和消費(fèi)者。
在這篇文章中，讓我們來看看如何實(shí)現(xiàn)這種非常有用的動態(tài)匹配模式。
首先，我們創(chuàng)建一個關(guān)于餐廳的Rust項(xiàng)目：
cargo init restaurant在Cargo.toml文件中加入依賴項(xiàng)：
[dependencies]
tokio = { version = "1.38.0", features = ["full"] }然后，在src/main.rs文件中寫入業(yè)務(wù)邏輯代碼。
作為餐廳經(jīng)理，可以分配不同的烹飪臺來異步準(zhǔn)備不同類型的食物，代碼如下(現(xiàn)在不用擔(dān)心未定義的值)：
async fn cooking_stand(food: char) {
loop {
somewhere.send(food.clone()).await;
}
}食物應(yīng)該被送到等待上菜的餐桌上，代碼如下：
async fn table (number: u8) {
loop {
let food = somehow.recv().await;
println!("Got {} at table {}", food, number);
}
}現(xiàn)在可以組織我們的餐廳了：
#[tokio::main]
async fn main() {
// 烹飪臺
tokio::spawn(cooking_stand('??')); // 沙拉
tokio::spawn(cooking_stand('??')); // 漢堡
// 客人餐桌
tokio::spawn(table(1));
tokio::spawn(table(2));
tokio::time::sleep(Duration::from_millis(1000)).await;
}為簡單起見，我們假設(shè)通過應(yīng)用程序接受訂單。例如，餐廳經(jīng)理(主線程)知道餐桌1正在等待沙拉，餐桌3正在等待漢堡。但如何真正完成這些訂單呢？
初級方法：
cooking_stand -> ?????????? -> -> table 1
cooking_stand -> ?????????? -> manager -> table 2
cooking_stand -> ?????????? -> -> table 3如果我們強(qiáng)迫經(jīng)理做這項(xiàng)工作，他可以等待沙拉烹飪站準(zhǔn)備沙拉，然后將其傳遞給餐桌1。然后等待漢堡烹飪臺準(zhǔn)備好漢堡，把它端到3號餐桌。
這顯然是一個有缺陷的設(shè)計(jì)：
- 不管是否需要，烹飪臺都會生產(chǎn)食物。
- 如果烹飪臺很慢，那么經(jīng)理將必須等待食物準(zhǔn)備好。
- 經(jīng)理不應(yīng)該做繁重的工作，因?yàn)檫@會影響他的反應(yīng)能力。
我們需要服務(wù)員，幸運(yùn)的是，Tokio為這項(xiàng)工作提供了完美的工具——oneshot 通道。這些通道被設(shè)計(jì)和優(yōu)化為一次傳遞單個值。
let (waiter_rx, waiter_tx) = oneshot::channel::<char>();為了讓服務(wù)員先把沙拉送到1號桌，需要修改我們的烹飪臺：
use tokio::sync::oneshot;
async fn cooking_stand (
product: char,
mut waiters: tokio::sync::mpsc::Receiver<oneshot::Sender<char>>
) {
while let Some(waiter) = waiters.recv().await {
waiter.send(product.clone());
}
}其中tokio::sync::mpsc::Receiver<oneshot::Sender<char>>是一個等待隊(duì)列。是的，你沒看錯，可以通過其他通道封裝一個oneshot通道。當(dāng)服務(wù)員到達(dá)烹飪臺時，烹飪臺就會把食物準(zhǔn)備好，然后交給服務(wù)員送到餐桌上。讓我們對餐桌做同樣的事情，他們有特定的服務(wù)員接收部分，會給他們送食物：
async fn table (
number: u8,
mut waiters: tokio::sync::mpsc::Receiver<oneshot::Receiver<char>>
) {
while let Some(waiter) = waiters.recv().await {
let food = waiter.await.unwrap();
println!("Got {} at table {}", food, number);
}
}當(dāng)服務(wù)員被分配到餐桌上時，顧客等待服務(wù)員送來烹飪臺生產(chǎn)的食物。為了完成這個謎題，我們來修改main函數(shù)。經(jīng)理可以雇傭服務(wù)員，而不是自己做繁重的工作，并將他們分配到匹配的烹飪臺和桌子上，以完成食物訂單。
#[tokio::main]
async fn main() {
// 經(jīng)理分配服務(wù)員到烹飪臺
let (stand_salad_tx, stand_salad_rx) = mpsc::channel::<oneshot::Sender<char>>(100);
let (stand_pizza_tx, stand_pizza_rx) = mpsc::channel::<oneshot::Sender<char>>(100);
let (stand_burger_tx, stand_burger_rx) = mpsc::channel::<oneshot::Sender<char>>(100);
// 搭建烹飪臺
tokio::spawn(cooking_stand('??', stand_salad_rx));
tokio::spawn(cooking_stand('??', stand_pizza_rx));
tokio::spawn(cooking_stand('??', stand_burger_rx));
// 經(jīng)理分配服務(wù)員到餐桌
let mut tables: Vec<tokio::sync::mpsc::Sender<oneshot::Receiver<char>>> = Vec::new();
for number in 1..=4 {
let (table_tx, table_rx) = mpsc::channel::<oneshot::Receiver<char>>(100);
tables.push(table_tx);
tokio::spawn(table(number, table_rx));
}
// t;
}讓我們通過在main函數(shù)的末尾添加以下代碼來檢查這種方式是否有效：
// 創(chuàng)建服務(wù)員
let (waiter_tx, waiter_rx) = oneshot::channel::<char>();
// 分配到沙拉烹飪臺
stand_salad_tx.send(waiter_tx).await.unwrap();
// 讓他把食物送到1號桌
tables.first().unwrap().send(waiter_rx).await.unwrap();運(yùn)行結(jié)果：
Got ?? at table 1這種通過常規(guī)通道發(fā)送oneshot通道的模式可以用于實(shí)現(xiàn)各種流量控制。以給定的比率、節(jié)流等方式傳遞消息。
完整代碼如下：
use std::time::Duration;
use tokio::sync::{mpsc, oneshot};
async fn cooking_stand(
product: char,
mut waiters: tokio::sync::mpsc::Receiver<oneshot::Sender<char>>,
) {
while let Some(waiter) = waiters.recv().await {
waiter.send(product).unwrap();
}
}
async fn table(number: u8, mut waiters: tokio::sync::mpsc::Receiver<oneshot::Receiver<char>>) {
while let Some(waiter) = waiters.recv().await {
let food = waiter.await.unwrap();
println!("Got {} at table {}", food, number);
}
}
#[tokio::main]
async fn main() {
// 經(jīng)理分配服務(wù)員到烹飪臺
let (stand_salad_tx, stand_salad_rx) = mpsc::channel::<oneshot::Sender<char>>(100);
let (stand_pizza_tx, stand_pizza_rx) = mpsc::channel::<oneshot::Sender<char>>(100);
let (stand_burger_tx, stand_burger_rx) = mpsc::channel::<oneshot::Sender<char>>(100);
// 搭建烹飪臺
tokio::spawn(cooking_stand('??', stand_salad_rx));
tokio::spawn(cooking_stand('??', stand_pizza_rx));
tokio::spawn(cooking_stand('??', stand_burger_rx));
// 經(jīng)理分配服務(wù)員到餐桌
let mut tables: Vec<tokio::sync::mpsc::Sender<oneshot::Receiver<char>>> = Vec::new();
for number in 1..=4 {
let (table_tx, table_rx) = mpsc::channel::<oneshot::Receiver<char>>(100);
tables.push(table_tx);
tokio::spawn(table(number, table_rx));
}
// 創(chuàng)建服務(wù)員
let (waiter_tx, waiter_rx) = oneshot::channel::<char>();
// 分配到沙拉烹飪臺
stand_salad_tx.send(waiter_tx).await.unwrap();
// 讓他把食物送到1號桌
tables.first().unwrap().send(waiter_rx).await.unwrap();
// 創(chuàng)建服務(wù)員
let (waiter_tx, waiter_rx) = oneshot::channel::<char>();
// 分配到披薩烹飪臺
stand_pizza_tx.send(waiter_tx).await.unwrap();
// 讓他把食物送到2號桌
tables.get(1).unwrap().send(waiter_rx).await.unwrap();
// 創(chuàng)建服務(wù)員
let (waiter_tx, waiter_rx) = oneshot::channel::<char>();
// 分配到披薩烹飪臺
stand_burger_tx.send(waiter_tx).await.unwrap();
// 讓他把食物送到3號桌
tables.get(2).unwrap().send(waiter_rx).await.unwrap();
tokio::time::sleep(Duration::from_millis(1000)).await;
}運(yùn)行結(jié)果：
Got ?? at table 2
Got ?? at table 3
Got ?? at table 1
