Csatornák a gyakorlatban: mpsc, oneshot és broadcast a Tokio-ban

Amikor több async task fut egymás mellett, előbb-utóbb szükség lesz arra, hogy kommunikáljanak egymással. Lehetne erre Arc<Mutex<T>>-et használni, de ez könnyen odavezet, hogy a lockok tartása közben blokkolódik az executor, vagy egyszerűen csak nehezen követhető lesz a kód. A csatornák (channel) ennek egy elegánsabb alternatívája: az egyik task küld egy üzenetet, a másik pedig .await-eli, amíg megérkezik. Nincs explicit lock, nincs megosztott mutálható állapot — csak üzenetek mozognak a taskok között.

A Tokio három fő csatornatípust ad a kezünkbe, mindegyik más kommunikációs mintára optimalizált:

• mpsc (multi-producer, single-consumer) — sok küldő, egy fogadó
• oneshot — pontosan egy üzenet, egyszeri válaszadásra
• broadcast — egy üzenet, sok fogadó, mindenki megkapja

Nézzük sorban, mikor melyiket érdemes használni.

Miért kell egyáltalán csatorna az async világban

Képzelj el egy taskot, amelyik egy TCP kapcsolatot kezel, és egy másikat, amelyik statisztikát gyűjt a bejövő kérésekről. A kapcsolatkezelő task minden feldolgozott kérés után szeretne jelezni a statisztikagyűjtőnek. Ha ezt megosztott állapoton keresztül oldanád meg, minden hozzáféréshez lock kellene, és a taskok közötti csatolás sokkal szorosabb lenne, mint amire szükség van.

A csatorna ezzel szemben egy tiszta határt húz: a küldő nem tudja, ki és hogyan olvassa az üzenetet, a fogadó nem tudja, honnan jött. Ez a fajta decoupling nagyon jól illik az async modellhez, ahol taskok gyakran függetlenül futnak, és csak alkalmanként kell szinkronizálódniuk.

Az mpsc csatorna: több küldő, egy fogadó

Ez a leggyakrabban használt csatornatípus. Több task is birtokolhat egy Sender-klónt, de csak egyetlen Receiver létezhet. Tipikus felhasználás: worker taskok eredményeket küldenek egy központi aggregátor felé.

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // A csatorna kapacitása 32 elem - ha ennyi üzenet vár
    // feldolgozásra, a send() blokkol (await-el), amíg hely nem lesz
    let (tx, mut rx) = mpsc::channel::<String>(32);

    for id in 0..3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            let msg = format!("worker {id} elkészült");
            if tx.send(msg).await.is_err() {
                eprintln!("a fogadó már nem figyel, worker {id} kilép");
            }
        });
    }

    // Az eredeti tx-et is le kell dobni, különben a csatorna
    // sosem záródik le, és a while let ciklus örökre vár
    drop(tx);

    while let Some(msg) = rx.recv().await {
        println!("kapott: {msg}");
    }

    println!("minden worker végzett, a csatorna lezárult");
}

Fontos apróság: a Sender klónozható (tx.clone()), a Receiver nem. Amikor az összes Sender-példány (beleértve a klónokat is) elpusztul, a recv() None-t ad vissza, és a while let ciklus természetesen kilép. Ez a lezárási mechanizmus az egyik legfontosabb dolog, amit meg kell érteni az mpsc-vel kapcsolatban.

A oneshot csatorna: egyszeri válaszküldés

Sokszor csak egyetlen válaszra van szükség egy taskból — tipikusan amikor egy task elindít egy műveletet egy másik taskban, és megvárja az eredményét. Ide illik pontosan a oneshot csatorna: pontosan egy üzenetet lehet rajta küldeni, aztán a csatorna lezárul.

use tokio::sync::oneshot;

async fn compute_answer() -> i32 {
    // valamilyen "nehéz" számítás szimulálása
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    42
}

#[tokio::main]
async fn main() {
    let (tx, rx) = oneshot::channel::<i32>();

    tokio::spawn(async move {
        let result = compute_answer().await;
        // a send() itt nem async - egyszeri, azonnali művelet
        let _ = tx.send(result);
    });

    match rx.await {
        Ok(value) => println!("megkaptam a választ: {value}"),
        Err(_) => eprintln!("a küldő task elszállt válasz nélkül"),
    }
}

Ez a minta rendkívül hasznos, amikor egy "request-response" jellegű kommunikációt kell megvalósítani két task között. Például egy háttérben futó adatbázis-worker task, amelyik mpsc-n kap request-eket, és mindegyikhez egy oneshot::Sender-t csatol a válaszhoz:

use tokio::sync::{mpsc, oneshot};

struct Request {
    payload: String,
    responder: oneshot::Sender<usize>,
}

async fn worker(mut rx: mpsc::Receiver<Request>) {
    while let Some(req) = rx.recv().await {
        let len = req.payload.len();
        // Nem baj, ha ez hibát ad - a kérő fél esetleg már nem vár
        let _ = req.responder.send(len);
    }
}

Ezzel a kombinációval egy tiszta, actor-szerű mintát kapunk: az mpsc a request-eket sorba állítja, a oneshot pedig gondoskodik az egyedi válaszok visszajuttatásáról a megfelelő hívóhoz.

A broadcast csatorna: egy üzenet, sok fogyasztó

Az mpsc-vel szemben, ahol egy üzenetet csak egyetlen fogadó kap meg, a broadcast csatornánál minden feliratkozó (Receiver) megkapja ugyanazt az üzenetet. Ez tipikusan olyan esetekben hasznos, mint egy chat szerver, ahol egy üzenetet minden csatlakozott klienshez el kell juttatni, vagy egy konfigurációváltozás jelzése több komponens felé.

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, _rx) = broadcast::channel::<&'static str>(16);

    let mut rx1 = tx.subscribe();
    let mut rx2 = tx.subscribe();

    tokio::spawn(async move {
        while let Ok(msg) = rx1.recv().await {
            println!("[fogyasztó 1] {msg}");
        }
    });

    tokio::spawn(async move {
        while let Ok(msg) = rx2.recv().await {
            println!("[fogyasztó 2] {msg}");
        }
    });

    tx.send("szerver újraindul 5 perc múlva").unwrap();
    tx.send("karbantartás elkezdődött").unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}

Minden subscribe() hívás egy új Receiver-t hoz létre, amely a hívás pillanatától kezdve kapja meg az üzeneteket — a korábbi üzeneteket nem látja. Ha egy fogyasztó lassabban olvas, mint amilyen sebességgel az üzenetek érkeznek, és lemarad a pufferméretnél is jobban, a recv() egy RecvError::Lagged(n) hibát ad vissza, jelezve, hogy n üzenetet elvesztett.

Backpressure és a csatorna kapacitása

Az mpsc és broadcast csatornák létrehozásakor meg kell adni egy kapacitást (channel(32)), ami a puffer méretét jelenti. Ez nem véletlen — ez a backpressure mechanizmus alapja. Ha a küldő gyorsabban produkál üzeneteket, mint amit a fogadó fel tud dolgozni, és a puffer betelik:

• mpsc esetén a send().await blokkol (async módon vár), amíg a fogadó fel nem szabadít helyet.
• broadcast esetén a régi üzenetek egyszerűen felülíródnak, a lassú fogyasztók pedig lemaradásjelzést kapnak.

A megfelelő kapacitás megválasztása gyakran kísérletezés kérdése: túl kicsi puffer felesleges blokkolást okoz, túl nagy puffer pedig memóriát pazarolhat, és elrejtheti azt a tényt, hogy a fogadó oldal lassabb, mint kellene.

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<u32>(2);

    tokio::spawn(async move {
        for i in 0..5 {
            println!("küldés előtt: {i}");
            tx.send(i).await.unwrap();
            println!("küldve: {i}");
        }
    });

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    while let Some(v) = rx.recv().await {
        println!("fogadva: {v}");
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
}

Ha lefuttatod ezt a kódot, jól látod, hogy a küldő task a harmadik üzenettől kezdve várakozni kezd, mert a puffer (kapacitás: 2) betelt, és a fogadó csak lassan üríti azt.

Gyakori hibák: lezárás és üzenetvesztés

A csatornákkal dolgozva néhány hiba rendszeresen visszatér:

1. Elfelejtett sender-klón életben tartása. Ha egy Sender-klónt véletlenül egy struct mezőjében tárolsz, ami sosem pusztul el, a csatorna sosem záródik le, és a fogadó örökre vár a recv()-en.

2. unwrap() a send() eredményén, ahol a fogadó közben eltűnhet. Ha a Receiver-t eldobják (pl. a task, amely olvasta, panic-olt vagy visszatért), a send() Err-t ad vissza. Ezt mindig kezelni kell, különben a küldő task is elszáll.

3. oneshot::Sender elfelejtett meghívása. Ha a tx.send()-et sosem hívod meg (pl. egy korai return miatt), a rx.await Err(RecvError)-t kap, amint a Sender drop-olódik. Ez let-else mintával könnyen kezelhető:

async fn handle(rx: tokio::sync::oneshot::Receiver<i32>) {
    let Ok(value) = rx.await else {
        eprintln!("a válasz sosem érkezett meg");
        return;
    };
    println!("feldolgozott érték: {value}");
}

4. Broadcast csatornán a Lagged hiba figyelmen kívül hagyása. Ha csak Ok(msg)-re mintázol a recv() eredményén, és a Lagged esetet nem kezeled, a ciklusod egyszerűen leáll az első lemaradásnál, holott a csatorna továbbra is működne.

Összefoglalás

A három csatornatípus mindegyike egy konkrét kommunikációs mintára van szabva: az mpsc a sok-küldő-egy-fogadó munkaelosztásra, a oneshot az egyszeri request-response párra, a broadcast pedig az egy-üzenet-sok-fogyasztó jelzésekre. A kulcs mindig az, hogy tisztán lásd, hány küldő és hány fogadó van a rendszeredben, és ennek megfelelően válaszd a csatornatípust. Ha ezt eltalálod, a taskjaid közötti kommunikáció olvasható, biztonságos és — a Rust type systemének hála — fordításidőben ellenőrzött lesz, nem futásidőben derül ki, hogy elfelejtettél valamit lezárni.