Csatornák Rustban: mpsc, oneshot és broadcast a gyakorlatban

Ha async Rustot írsz, egy ponton biztosan szembejön a kérdés: hogyan üzenjenek egymásnak a task-ok anélkül, hogy Mutex-eket dobálnál mindenhova? A válasz szinte mindig valamilyen csatorna. A tokio::sync modul három fő típust ad a kezünkbe: mpsc, oneshot és broadcast. Ezek mindegyike más problémára optimális, és pontosan ez a cikk célja: megmutatni, melyiket mikor húzd elő a szerszámosládából.

Mikor melyik csatornát válasszuk?

Mielőtt belevágnánk a kódba, tisztázzuk a döntési fát:

• mpsc (multi-producer, single-consumer): sok task küld üzenetet, egy task fogadja. Ez a leggyakoribb minta – gondolj egy worker pool-ra, ahol a munkát sorba állítod.
• oneshot: egyetlen üzenet, egyetlen küldő, egyetlen fogadó. Tipikusan akkor kell, amikor egy task elindít egy másikat, és vár egy "eredményre" – ez a request-response minta async világban.
• broadcast: egy küldő, sok fogadó, és mindenki megkapja az összes üzenetet. Ilyenkor gondolj egy event bus-ra, ahol több komponens is figyeli ugyanazt az eseményfolyamot.

Fontos: ezek nem versenyeznek egymással, hanem kiegészítik egymást. Egy komolyabb async rendszerben simán előfordul, hogy mindhármat használod egyszerre.

mpsc: a klasszikus producer-consumer

Az mpsc::channel a leggyakrabban használt csatorna típus. A channel(buffer_size) függvény egy (Sender<T>, Receiver<T>) páros ad vissza. A Sender klónozható, tehát több task is küldhet rajta, de csak egy Receiver létezhet.

use tokio::sync::mpsc;

#[derive(Debug)]
struct Job {
    id: u32,
    payload: String,
}

async fn worker(mut rx: mpsc::Receiver<Job>) {
    while let Some(job) = rx.recv().await {
        println!("Feldolgozás: {} -> {}", job.id, job.payload);
    }
    println!("Nincs több küldő, a worker leáll.");
}

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

    let handle = tokio::spawn(worker(rx));

    for i in 0..5 {
        let tx = tx.clone();
        tokio::spawn(async move {
            let job = Job { id: i, payload: format!("munka-{i}") };
            if tx.send(job).await.is_err() {
                eprintln!("A receiver már bezárt");
            }
        });
    }

    // Fontos: a fő tx-et is drop-oljuk, különben a receiver sosem áll le
    drop(tx);

    handle.await.unwrap();
}

Ami itt kulcsfontosságú: a Receiver::recv() None-t ad vissza, amikor minden Sender klón elszáll (drop-olódik). Ez az a pont, ahol a legtöbb kezdő elakad – ha az eredeti tx-et nem drop-olod, a worker sosem lát None-t, és a program örökre a recv().await-en fog várni.

oneshot: egyszeri válaszok

A oneshot::channel pontosan egy értéket enged átküldeni, aztán a csatorna "elhasználódik". Ez tökéletes arra, amikor egy task elindít egy műveletet egy másik task-on, és megvárja az eredményt – klasszikus request-response minta, csak async formában.

use tokio::sync::oneshot;

async fn compute_answer(tx: oneshot::Sender<i32>) {
    let result = 21 * 2;
    // A send() nem async, azonnal visszatér
    let _ = tx.send(result);
}

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

    tokio::spawn(compute_answer(tx));

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

Ez a minta rendkívül hasznos, ha van egy szervíz task-od (pl. egy adatbázis-kapcsolatot kezelő worker), amely mpsc-n kap parancsokat, és minden parancshoz egy oneshot::Sender-t csatolsz, hogy a válasz pontosan visszajusson a hívóhoz.

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

enum Command {
    Get { key: String, resp: oneshot::Sender<Option<String>> },
}

async fn store_actor(mut rx: mpsc::Receiver<Command>) {
    let mut data = std::collections::HashMap::new();
    data.insert("foo".to_string(), "bar".to_string());

    while let Some(cmd) = rx.recv().await {
        match cmd {
            Command::Get { key, resp } => {
                let value = data.get(&key).cloned();
                let _ = resp.send(value);
            }
        }
    }
}

broadcast: több feliratkozóval

Ha több task-nak is szüksége van ugyanarra az információra – például egy konfigurációváltozás eseményére, vagy egy "shutdown" jelzésre –, a broadcast::channel a barátod. Minden Receiver, amit a subscribe() hívással kapsz, megkapja az összes üzenetet a saját feliratkozása pillanatától kezdve.

use tokio::sync::broadcast;

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

    for id in 0..3 {
        let mut rx = tx.subscribe();
        tokio::spawn(async move {
            loop {
                match rx.recv().await {
                    Ok(msg) => println!("[feliratkozó {id}] üzenet: {msg}"),
                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        eprintln!("[feliratkozó {id}] {n} üzenetet elveszítettünk!");
                    }
                    Err(broadcast::error::RecvError::Closed) => break,
                }
            }
        });
    }

    tx.send("szerver újraindul".to_string()).unwrap();
    tx.send("konfiguráció frissült".to_string()).unwrap();

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

Érdemes megjegyezni, hogy a broadcast csatornának is van buffer mérete, de ez másképp viselkedik, mint az mpsc-nél: ha egy feliratkozó lassabb, mint az üzenetek érkezési sebessége, és a buffer megtelik, a legrégebbi üzeneteket a csatorna eldobja az adott feliratkozó számára, és a RecvError::Lagged(n) hibát kapod meg n elveszett üzenettel. Ez nem crash, csak figyelmeztetés – neked kell eldönteni, mit teszel vele.

Backpressure és buffer méret kezelése

A buffer méret megválasztása nem csak technikai apróság, hanem architekturális döntés. Nézzük a lehetőségeket:

• Kis buffer (pl. 1-8): a send().await gyakran blokkol, amíg a fogadó fel nem szabadít helyet. Ez erős backpressure-t jelent – a küldő task lelassul, ha a fogadó nem tud lépést tartani. Ez sokszor kívánatos, mert megakadályozza, hogy a memória felduzzadjon.
• Nagy buffer: rugalmasabb, kevesebb blokkolás, de ha a fogadó tartósan lemarad, a memóriahasználat elszabadulhat, és az egész rendszer instabillá válhat.

Egy gyakori hiba, hogy valaki mpsc::channel(10000)-et választ "biztonság kedvéért", majd csodálkozik, amikor a program megeszik fél gigabájt RAM-ot terhelés alatt. A helyes megközelítés: mérd meg, mekkora a tipikus terhelés, és válassz egy olyan buffer méretet, amely még kezelhető torlódást enged, de nem hagyja elszabadulni a memóriát.

use tokio::sync::mpsc;

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

    let producer = tokio::spawn(async move {
        for i in 0..10 {
            match tx.try_send(i) {
                Ok(_) => println!("elküldve: {i}"),
                Err(mpsc::error::TrySendError::Full(val)) => {
                    println!("telítve, kihagyva: {val}");
                }
                Err(mpsc::error::TrySendError::Closed(_)) => break,
            }
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
    });

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

    producer.await.unwrap();
}

Ebben az esetben a producer gyorsabb, mint a consumer, és a try_send segítségével explicit módon dönt arról, mi történjen, ha nincs hely – eldobja az üzenetet, helyette akár logolhat, metrikát növelhet, vagy másik stratégiát is választhatna (pl. várakozás helyett egy másik csatornára irányítás).

Összefoglalás

Ha összefoglalnám egy mondatban: mpsc a munkaelosztásra, oneshot a válaszokra, broadcast az értesítésekre. A három csatorna típus kombinálásával rendkívül tiszta, actor-szerű architektúrákat tudsz felépíteni Rust async kódban, Mutex és megosztott állapot nélkül. A buffer méret és a backpressure kezelése pedig nem mellékes részlet – ez dönti el, hogy a rendszered kontrolláltan lassul le terhelés alatt, vagy egyszerűen felfalja a memóriát. Következő lépésként érdemes megnézni a tokio::sync::watch csatornát is, amely egy speciális eset: csak a legutolsó értéket tartja meg, ami konfigurációk vagy állapot-jelzők terjesztésére ideális – de ez már egy másik cikk témája.