Csatornák Rust async világában: mpsc, oneshot és broadcast összehasonlítva

Amikor async kódot írsz, előbb-utóbb szembejön a probléma: két vagy több task között üzenetet kell átadni. A megosztott Mutex<T> sokszor túlbonyolítja a dolgot, és könnyen deadlockba fut, ha nem vagy elég figyelmes. Itt jönnek képbe a csatornák (channel), amik message passing-alapú kommunikációt tesznek lehetővé - pont úgy, ahogy Go-ban a channel, csak Rust módra, típusbiztosan.

A cikkben végigmegyünk a leggyakoribb csatornatípusokon: az std könyvtárbeli mpsc-n, a tokio mpsc, oneshot és broadcast csatornáin. Mindegyiknek megvan a maga jól körülhatárolt use case-e, és ha rosszul választasz, könnyen felesleges overheadet vagy pont hiányzó funkciót kapsz.

Mikor melyik csatornát válasszuk?

Mielőtt belevágnánk a kódba, érdemes tisztázni, hogy a különböző csatornatípusok milyen kommunikációs mintát fednek le:

• mpsc (multi-producer, single-consumer): sok küldő, egy fogadó. Ez a leggyakoribb minta - például worker taskok küldik az eredményeiket egy központi feldolgozónak.
• oneshot: pontosan egy üzenet, pontosan egyszer. Tipikusan akkor kell, ha egy taskot elindítasz, és várod, hogy visszaadjon egy eredményt - olyasmi, mint egy async "function call", csak task-határon átívelve.
• broadcast (multi-producer, multi-consumer): minden fogadó megkapja az összes üzenetet. Ez kell, ha több független komponensnek is értesülnie kell ugyanarról az eseményről.

Az std könyvtár csak mpsc-t ad, a oneshot és broadcast a tokio saját fejlesztése - ezek nem véletlenül async-specifikusak, hiszen aszinkron várakozást (.await) implementálnak a recv() hívásban.

Az std::sync::mpsc újjászületése Rust 1.67-ben

Ha régóta Rust-ozol, biztos ismered a std::sync::mpsc-t. Ez a szinkron, blokkoló csatorna típus, ami szálak (nem async taskok!) közötti kommunikációra való. Az elmúlt években viszont az implementációja elég régi és lassú volt - a belső algoritmus egy komplex, több queue-t kombináló megoldás volt, ami nem öregedett túl jól.

A Rust 1.67-es kiadásában a standard könyvtár csapata teljesen újraírta a std::sync::mpsc belső implementációját. Az új verzió egyszerűbb, gyorsabb, és jelentősen csökkenti a lock contention-t nagy terhelés mellett. A publikus API ugyanaz maradt, tehát a kódodban semmit nem kell módosítanod - csak gyorsabban fog futni.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for i in 0..4 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(i * i).expect("küldés sikertelen");
        });
    }
    drop(tx);

    let mut results: Vec<i32> = rx.iter().collect();
    results.sort();
    println!("{:?}", results);
}

Fontos hangsúlyozni: ez a mpsc nem async, hanem szinkron szálak közti csatorna. Ha .await-elhető csatornára van szükséged (tehát async taskok között akarsz kommunikálni), ne ezt használd - arra való a tokio::sync::mpsc.

tokio::sync::mpsc gyakorlatban

Aszinkron kontextusban a tokio::sync::mpsc a natúr választás. Az API hasonlít az std megfelelőjére, de a send és recv metódusok .await-elhetők, és a csatorna korlátozott kapacitású (bounded), ami backpressure-t biztosít - ha a fogyasztó lassú, a küldők várni fognak, amíg hely nem szabadul fel.

[dependencies]
tokio = { version = "1", features = ["full"] }

use tokio::sync::mpsc;

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

    for i in 0..5 {
        let tx = tx.clone();
        tokio::spawn(async move {
            if tx.send(i).await.is_err() {
                eprintln!("a fogadó már nem figyel");
            }
        });
    }
    drop(tx);

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

Ami itt lényeges: a tx.clone() minden worker taskban lehetővé teszi, hogy mind az öt task ugyanabba a csatornába küldjön. Az eredeti tx-et a drop(tx) sorral elengedjük a main-ben, különben a while let Some(...) sosem térne vissza None-nal - a csatorna addig van "élve", amíg legalább egy Sender létezik.

A bounded csatorna kapacitása (mpsc::channel::<u32>(32)) azért fontos, mert ha nincs korlát, egy gyors küldő és lassú fogyasztó kombinációja korlátlanul növelheti a memóriahasználatot. Ha valóban korlátlan pufferre van szükséged (ritkán indokolt), a tokio::sync::mpsc::unbounded_channel függvényt használhatod, de ezzel óvatosan bánj.

oneshot: egyszeri válaszok csatornája

Gyakori minta, hogy elindítasz egy async taskot, és csak egyetlen eredményre vársz tőle - mintha egy függvényhívás lenne, csak külön taskban fut. Erre a legjobb eszköz a tokio::sync::oneshot::channel.

use tokio::sync::oneshot;

async fn compute() -> u64 {
    42
}

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

    tokio::spawn(async move {
        let result = compute().await;
        let _ = tx.send(result);
    });

    let Ok(value) = rx.await else {
        eprintln!("a küldő oldal eltűnt, mielőtt válaszolt volna");
        return;
    };

    println!("az eredmény: {value}");
}

Itt egy let-else konstrukciót is bevetettünk - ez nagyon jól illik az ilyen "vagy sikerül, vagy korai kilépés" mintákhoz, sokkal olvashatóbb, mint egy beágyazott match.

A oneshot::Sender::send nem async - azonnal visszatér, mert a oneshot csatorna kapacitása pontosan egy elem, és nincs szükség várakozásra a küldő oldalon. Ha a fogadó (Receiver) már el lett dobva, a send egyszerűen Err-t ad vissza az elküldeni kívánt értékkel együtt, amit itt figyelmen kívül hagyunk a let _ = mintával.

broadcast: egy üzenet, sok fogyasztó

Ha több független komponensnek is meg kell kapnia ugyanazt az üzenetet - például egy eseményrendszerben, ahol több listener figyel egy állapotváltozásra -, a tokio::sync::broadcast a megoldás. Minden subscribe()-bal létrehozott Receiver megkapja a küldött üzenetek másolatát.

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, mut rx1) = broadcast::channel(16);
    let mut rx2 = tx.subscribe();

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

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

    for i in 0..3 {
        tx.send(format!("üzenet #{i}")).unwrap();
    }

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

Fontos részlet: a broadcast csatorna is bounded, itt a 16 a belső ringbuffer méretét jelenti. Ha egy lassú fogyasztó nem elég gyorsan olvassa ki az üzeneteket, és a buffer megtelik, a recv() Err(RecvError::Lagged(n)) hibát ad vissza, jelezve, hogy n üzenetet kihagyott. Ez tudatos tervezési döntés - a broadcast inkább "legfrissebb állapot" jellegű adatokhoz illik, nem garantált, veszteségmentes kézbesítéshez.

Összefoglalás

A csatornák az async Rust egyik legfontosabb, legjobban bevált eszközei, és a döntés valójában egyszerű, ha tisztázod a kommunikációs mintát:

• Szinkron szálak között, sok küldő és egy fogadó: std::sync::mpsc - és jó hír, hogy Rust 1.67 óta ez alattad is gyorsabb lett anélkül, hogy bármit módosítanod kellett volna.
• Async taskok között, sok küldő és egy fogadó: tokio::sync::mpsc, backpressure-rel.
• Egyszeri válasz egy tasktól: tokio::sync::oneshot, minimális overhead-del.
• Egy üzenet, sok fogyasztónak: tokio::sync::broadcast, tudatosan vállalva a lehetséges lag-et.

Ha ezeket a mintákat helyesen alkalmazod, sokkal olvashatóbb és biztonságosabb lesz a konkurrens kódod, mint ha megosztott Mutex-ekkel próbálnád meg ugyanazt megoldani. Próbáld ki mindegyiket egy kis teszt projektben - a legjobb tanulás mindig a gyakorlat.