Async Rust mit Tokio: async/await, task und channel

1. Übersicht

Rust hat eine eingebaute async/await-Syntax, bringt aber keine Runtime zum Ausführen von Future mit sich. Tokio ist die beliebteste Runtime dafür. Sie bietet einen Scheduler, Timer, asynchrone I/O und Synchronisierungswerkzeuge.

Der Artikel führt von der ersten async-Funktion zu häufig verwendeten Operationen: Task spawnen, paralleles Ausführen mit join!, Timeout mit select! setzen, Zustand zwischen Tasks teilen, Daten über Channel senden und Blocking-Code verarbeiten. Der letzte Abschnitt behandelt das Schreiben von Tests und häufige Fehler.

2. Installation

Sie fügen Tokio zu Cargo.toml hinzu:

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

[dev-dependencies]
tokio = { version = "1", features = ["full", "test-util"] }

Das Feature full aktiviert alle Module: Multi-Thread-Runtime, time, sync, net, macros… Diese Methode ist beim Lernen bequem. In echten Projekten sollten Sie nur die Features aktivieren, die Sie wirklich verwenden, um die Kompilierungszeit zu verkürzen, beispielsweise ["rt-multi-thread", "macros", "time", "sync"]. Das Feature test-util ermöglicht es, die Zeit in Tests zu steuern. Wir werden es in Abschnitt 8 verwenden.

3. Die erste async-Funktion

3.1. #[tokio::main] und .await

use std::time::Duration;
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100)).await; // Datenbankaufruf simulieren
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let order = fetch_order(42).await;
    println!("{order}");
}

Erklärung:

• async fn fetch_order gibt nicht direkt String zurück. Sie gibt ein impl Future<Output = String> zurück.
• tokio::time::sleep ist die asynchrone Variante von sleep. Während des Wartens wird der Thread an die Runtime zurückgegeben, damit diese andere Aufgaben ausführen kann.
• #[tokio::main] ist ein Makro. Es erstellt eine Multi-Thread-Runtime und führt dann main darin mittels block_on aus.

Order #42

3.2. Future ist lazy

Das ist ein großer Unterschied zu Promises in JavaScript: Futures in Rust tun nichts, bis Sie sie mit .await ansteuern oder sie gepollt werden.

async fn load_customer(id: u32) -> String {
    println!("Lade Customer {id}");
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let fut = load_customer(7); // noch keine Zeile in der Funktion wird ausgeführt
    println!("Vor dem await");
    let customer = fut.await;
    println!("{customer}");
}

Vor dem await
Lade Customer 7
Customer 7

Die Zeile “Lade Customer 7” erscheint erst nach dem .await-Aufruf. Wenn Sie .await vergessen, wird die Funktion nie ausgeführt. Der Compiler warnt Sie mit unused implementer of Future that must be used, daher ignorieren Sie diese Warnung bitte nicht.

4. Task spawnen

.await führt Futures sequenziell aus. Um sie gleichzeitig auszuführen, verwenden Sie tokio::spawn, um eine Task zu erstellen. Eine Task ist ein leichter “Green Thread”, den Tokio verwaltet.

use std::time::{Duration, Instant};
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100 * id as u64)).await;
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let mut handles = Vec::new();

    for id in 1..=3 {
        handles.push(tokio::spawn(async move { fetch_order(id).await }));
    }

    for handle in handles {
        let order = handle.await.expect("Task panicked");
        println!("{order}");
    }
    println!("Gesamtzeit: ~{} ms", start.elapsed().as_millis());
}

Order #1
Order #2
Order #3
Gesamtzeit: ~300 ms

Wenn Sie sequenziell ausführen würden, würde die Gesamtzeit 100 + 200 + 300 = 600 ms betragen. Beim Spawning laufen die drei Tasks gleichzeitig, daher dauert es insgesamt nur etwa 300 ms.

Einige wichtige Punkte:

• tokio::spawn gibt JoinHandle<T> zurück. Wenn Sie den Handle mit .await ansteuern, erhalten Sie Result<T, JoinError>. Das Ergebnis ist Err, wenn die Task panicked oder abgebrochen wird.
• Das übergebene Future muss Send + 'static erfüllen, da die Task auf einem anderen Thread laufen kann und länger als der aktuelle Scope existieren könnte. Deshalb benötigen Sie async move.
• Tasks starten sofort nach dem Spawning, ohne auf das .await des Handles zu warten.

Hinweis: Das Löschen eines JoinHandle bricht die Task nicht ab. Die Task läuft im Detached-Modus weiter. Um sie abzubrechen, rufen Sie handle.abort() auf.

Wenn sich die Anzahl der Tasks ändert, können Sie tokio::task::JoinSet verwenden. Es ermöglicht es, Ergebnisse nach der Reihenfolge abzurufen, in der Tasks mit join_set.join_next().await abgeschlossen werden.

5. Gleichzeitiges Ausführen in derselben Task: join!, select!, timeout

5.1. join! und try_join!

Sie müssen nicht immer spawnen. tokio::join! pollt mehrere Futures gleichzeitig in derselben Task und wartet, bis alle abgeschlossen sind:

use std::time::{Duration, Instant};
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

async fn fetch_customer(id: u32) -> String {
    sleep(Duration::from_millis(150)).await;
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let (order, customer) = tokio::join!(fetch_order(1), fetch_customer(7));
    println!("{order} - {customer} ({} ms)", start.elapsed().as_millis());
}

Order #1 - Customer 7 (200 ms)

Die Gesamtzeit entspricht dem längsten Future, etwa 200 ms. Die Futures müssen nicht 'static sein, daher können sie lokale Variablen ausleihen.

Wenn die Funktionen Result zurückgeben, verwenden Sie tokio::try_join!. Dieses Makro stoppt beim ersten Fehler und verwirft die verbleibenden Futures:

let (order, customer) = tokio::try_join!(fetch_order_checked(1), fetch_customer_checked(7))?;

Hier sind fetch_order_checked und fetch_customer_checked async-Funktionen, die Result<String, AppError> zurückgeben.

5.2. timeout

use tokio::time::{timeout, Duration};

match timeout(Duration::from_millis(50), fetch_order(1)).await {
    Ok(order) => println!("{order}"),
    Err(_) => println!("Zeitlimit überschritten"),
}

Zeitlimit überschritten

fetch_order benötigt 200 ms, was über dem Limit von 50 ms liegt. In diesem Fall gibt timeout Err(Elapsed) zurück, und das Future darin wird gelöscht, also abgebrochen.

5.3. select!

select! wartet auf das erste abgeschlossene Future und bricht dann die restlichen ab:

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    tokio::spawn(async move {
        sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(());
    });

    tokio::select! {
        order = fetch_order(1) => println!("Bestellung verarbeitet {order}"),
        _ = shutdown_rx => println!("Shutdown-Signal erhalten, verarbeitung stoppen"),
    }
}

Shutdown-Signal erhalten, verarbeitung stoppen

Das Shutdown-Signal kommt nach 100 ms, früher als fetch_order (200 ms), daher gewinnt der zweite Zweig.

Hinweis: Abbrechen in Rust bedeutet, dass das Future an der nächsten .await-Stelle gelöscht wird. Wenn der abgebrochene Zweig eine nicht-atomare Operation durchführt, beispielsweise hat bereits die Hälfte einer Message aus einem Socket gelesen, können Daten verloren gehen. Sie sollten die “cancel safety”-Dokumentation jeder Funktion überprüfen, bevor Sie sie in select! in einer Schleife verwenden.

6. Daten zwischen Tasks teilen

6.1. Arc<Mutex<T>>

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

#[tokio::main]
async fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("laptop", 10u32)])));
    let mut handles = Vec::new();

    for _ in 0..5 {
        let inventory = Arc::clone(&inventory);
        handles.push(tokio::spawn(async move {
            let mut stock = inventory.lock().unwrap();
            *stock.get_mut("laptop").unwrap() -= 1;
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    println!("Verbleibend: {}", inventory.lock().unwrap()["laptop"]);
}

Verbleibend: 5

Hier verwenden wir std::sync::Mutex, nicht tokio::sync::Mutex. Dies ist die empfohlene Wahl, wenn der kritische Bereich kurz ist und kein .await darin vorhanden ist, da der Standard-Mutex schneller ist.

Sie benötigen nur tokio::sync::Mutex, wenn Sie einen Lock über einen .await-Punkt hinweg halten müssen, beispielsweise wenn Sie eine gemeinsam genutzte Verbindung während des Wartens auf I/O halten.

6.2. Channel mpsc

Anstatt Speicher zu teilen, können Sie Nachrichten zwischen Tasks über einen Channel übertragen:

use tokio::sync::mpsc;

#[derive(Debug)]
struct Order {
    id: u32,
    amount: u64,
}

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

    for id in 1..=3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(Order { id, amount: id as u64 * 100 }).await.unwrap();
        });
    }
    drop(tx); // Sender schließen, damit die Schleife beendet wird

    let mut total = 0;
    while let Some(order) = rx.recv().await {
        println!("Erhalten {order:?}");
        total += order.amount;
    }
    println!("Gesamt: {total}");
}

Erhalten Order { id: 1, amount: 100 }
Erhalten Order { id: 3, amount: 300 }
Erhalten Order { id: 2, amount: 200 }
Gesamt: 600

Die Reihenfolge der Nachrichten kann sich zwischen den Durchläufen unterscheiden. Der Parameter 32 ist die Puffergröße. Wenn der Buffer voll ist, wartet send. recv() gibt None zurück, wenn alle Sender gelöscht wurden. Wenn Sie drop(tx) vergessen, wird das Programm endlos in der Schleife steckenbleiben.

Neben mpsc hat Tokio auch:

• oneshot: eine einzelne Nachricht senden.
• broadcast: mehrere Consumer, jeder empfängt jede Nachricht.
• watch: Consumer interessieren sich nur für den neuesten Wert.

7. Handling von blockierendem Code

Tokio nutzt Cooperative Scheduling. Tasks geben den Thread nur an den Stellen .await ab. Eine CPU-intensive Operation oder synchrone I/O belegt den Worker-Thread komplett und verhungert andere Tasks.

fn hash_password(password: &str) -> String {
    // Simuliert eine CPU-intensive Operation, z.B. bcrypt/argon2
    std::thread::sleep(std::time::Duration::from_millis(500));
    format!("hashed:{password}")
}

#[tokio::main]
async fn main() {
    let password = String::from("s3cret");
    let hashed = tokio::task::spawn_blocking(move || hash_password(&password))
        .await
        .expect("Task mit Blocking panicked");
    println!("{hashed}");
}

hashed:s3cret

spawn_blocking schiebt die Closure in einen separaten Thread-Pool für Blocking-Aufgaben. Dadurch bleiben die Worker-Threads der Runtime frei, um andere Requests zu verarbeiten.

Hinweis: Rufen Sie nicht std::thread::sleep, std::fs::read oder einen synchronen HTTP-Client direkt in einer async-Funktion auf. Verwenden Sie die async-Variante (tokio::time::sleep, tokio::fs) oder wickeln Sie sie in spawn_blocking. Für massives paralleles Computing ist Rayon oft besser geeignet.

8. Testen von async-Code

Das Makro #[tokio::test] erstellt für jeden Test eine Runtime. Standardmäßig ist dies eine Single-Thread-Runtime.

use std::time::Duration;
use tokio::time::{sleep, timeout};

pub async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn fetch_order_returns_formatted_id() {
        assert_eq!(fetch_order(42).await, "Order #42");
    }

    #[tokio::test(start_paused = true)]
    async fn timeout_fires_when_order_service_is_slow() {
        let result = timeout(Duration::from_millis(50), fetch_order(1)).await;
        assert!(result.is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn order_arrives_within_generous_timeout() {
        let result = timeout(Duration::from_secs(1), fetch_order(1)).await;
        assert_eq!(result.unwrap(), "Order #1");
    }
}

cargo test

running 3 tests
test tests::fetch_order_returns_formatted_id ... ok
test tests::timeout_fires_when_order_service_is_slow ... ok
test tests::order_arrives_within_generous_timeout ... ok

start_paused = true (benötigt das Feature test-util) friert die Tokio-Uhr ein. Wenn alle Tasks auf einen Timer warten, springt die Runtime automatisch zur nächsten Zeitmarke. Dadurch laufen Timeout-Tests quasi sofort und liefern stabile Ergebnisse, unabhängig von der CI-Maschinengeschwindigkeit.

Wenn Sie das Verhalten auf mehreren Threads überprüfen müssen, verwenden Sie #[tokio::test(flavor = "multi_thread", worker_threads = 2)].

9. Häufige Fehler

Ein std::sync::MutexGuard über .await halten. Dieser Guard ist nicht Send, daher ist das Future auch nicht mehr Send. tokio::spawn gibt dann einen Compilerfehler future cannot be sent between threads safely. Die Lösung ist, den Guard-Scope mit einem Block { ... } vor dem .await zu verkürzen, oder zu tokio::sync::Mutex zu wechseln.

Geschachtelte Runtimes erstellen. Wenn Sie Runtime::new().block_on(...) innerhalb von Code aufrufen, der bereits auf Tokio läuft, panicked es mit der Meldung Cannot start a runtime from within a runtime. Im async-Kontext reicht einfach .await.

Async verwenden, wenn Sie es nicht brauchen. Ein CLI-Tool, das ein paar Dateien liest, oder ein Job mit reiner Rechenarbeit profitiert nichts von Tokio. Sie zahlen nur zusätzliche Kompilierungszeit und Komplexität. Async zahlt sich aus, wenn viele I/O-Operationen gleichzeitig anfallen: Web-Server, Crawler, Proxy, Client, der mehrere Services aufruft.

Unbegrenztes Spawning. Tasks sind billig, aber nicht kostenlos. Eine Task für jedes Element in einer Liste mit einer Million Elementen kann Downstream-Ressourcen erschöpfen, wie den Connection Pool einer Datenbank. Begrenzen Sie die Parallelität mit tokio::sync::Semaphore oder verwenden Sie buffer_unordered aus der Crate futures.

JoinError ignorieren. Wenn Sie den Handle mit .await ansteuern und let _ = verwenden, wird der Panic in der Task verschluckt. Sie sollten den Fehler loggen oder weitergeben.

10. Zusammenfassung

Der Artikel hat die Kernthemen von async Rust mit Tokio durchgenommen:

• Futures sind lazy und laufen nur, wenn sie .await angesteuert werden.
• tokio::spawn erstellt gleichzeitig laufende Tasks. Das übergebene Future muss Send + 'static sein.
• join! und try_join! führen mehrere Futures in derselben Task aus. select! und timeout handhaben Rennbedingungen und Abbrüche.
• Gemeinsam genutzte Daten können über Arc<Mutex<T>> geteilt oder über Channels wie mpsc und oneshot übertragen werden.
• Blockierender Code muss in spawn_blocking verschoben werden.
• #[tokio::test(start_paused = true)] macht Tests zur Laufzeit schnell und stabil.

Wenn Sie diese Themen beherrschen, können Sie mit asynchronem Netzwerk-I/O über tokio::net fortfahren oder Web-Services auf Tokio-Basis mit Frameworks wie Axum erstellen.