Reaktive Programmierung in Rust: Bibliotheken, Beispiele und Praxis

Reaktive Programmierung in Rust: Von beobachtbaren Objekten bis zu asynchronen Streams

\n\n

Reaktive Programmierung ist ein Paradigma, das auf die Verarbeitung asynchroner Datenströme und die Verbreitung von Änderungen ausgerichtet ist. Im Rust-Ökosystem wird es durch eine Kombination aus einem leistungsstarken Typsystem, speichersicherer Arbeit und modernen asynchronen Primitiven umgesetzt. In diesem Artikel betrachten wir die wichtigsten Bibliotheken, Muster und praktische Beispiele der reaktiven Programmierung in Rust.

\n\n

Wichtige Bibliotheken für reaktive Programmierung in Rust

\n\n

In Rust gibt es keinen einheitlichen „Standard“ für reaktive Programmierung, aber es existieren mehrere ausgereifte Bibliotheken, die verschiedene Abstraktionsebenen abdecken:

\n\n

\n
• RxRust – ein Port des beliebten ReactiveX (Rx) für Rust. Bietet Observable, Observer, Scheduler und eine breite Palette von Operatoren (map, filter, flat_map, merge usw.).
\n
• futures-rs – eine grundlegende Bibliothek für die Arbeit mit asynchronen Berechnungen. Das Stream-Trait ist die Grundlage für reaktive Streams.
\n
• async-std und tokio – asynchrone Laufzeiten, die Kanäle (channels), Timer und andere Primitive für reaktive Ketten bereitstellen.
\n
• frunk – eine Bibliothek für funktionale Programmierung, die reaktive Muster wie Lens und State enthält.
\n

\n\n

Beispiel 1: Einfaches Observable mit RxRust

\n\n

RxRust implementiert das „Beobachter“-Muster: Ein Observable erzeugt Ereignisse, ein Observer abonniert sie.

\n\n

// Cargo.toml: rxrust = "1.0"\nuse rxrust::prelude::*;\n\nfn main() {\n    let observable = observable::from_iter(vec![1, 2, 3, 4, 5]);\n\n    observable\n        .filter(|x| x % 2 == 0)\n        .map(|x| x * 10)\n        .subscribe(|x| println!("Erhalten: {}", x));\n}\n

\n\n

In diesem Beispiel wird ein Stream aus den Zahlen 1..5 erstellt, gerade Zahlen gefiltert, mit 10 multipliziert und ausgegeben. Die Operatorkette wird träge ausgeführt – erst nach dem Aufruf von subscribe.

\n\n

Beispiel 2: Asynchrone Streams mit futures-rs

\n\n

Die Bibliothek futures-rs bietet das Trait Stream, das asynchrone Gegenstück zu Iterator. Kombinatoren aus StreamExt ermöglichen den Aufbau reaktiver Ketten.

\n\n

// Cargo.toml: futures = "0.3"\nuse futures::stream::{self, StreamExt};\n\n#[tokio::main]\nasync fn main() {\n    let stream = stream::iter(0..10)\n        .filter(|x| futures::future::ready(x % 2 == 0))\n        .map(|x| x * 10)\n        .take(3);\n\n    let result: Vec<i32> = stream.collect().await;\n    println!("Ergebnis: {:?}", result); // [0, 20, 40]\n}\n

\n\n

Hier akzeptiert filter eine asynchrone Funktion, was es ermöglicht, Aufrufe externer APIs oder Datenbanken in die Kette einzubinden. take(3) begrenzt die Anzahl der Elemente – dies ist ein Beispiel für „Gegendruck“ (backpressure).

\n\n

Beispiel 3: Reaktive Kanäle mit async-std

\n\n

Kanäle sind eine klassische Methode zur Umsetzung reaktiver Interaktionen zwischen Komponenten. async-std bietet Mehrfachproduzenten-, Mehrfachkonsumenten-Kanäle.

\n\n

// Cargo.toml: async-std = "1.12"\nuse async_std::channel;\nuse async_std::task;\n\n#[async_std::main]\nasync fn main() {\n    let (sender, receiver) = channel::unbounded();\n\n    // Produzent\n    task::spawn(async move {\n        for i in 0..5 {\n            sender.send(i).await.unwrap();\n            task::sleep(std::time::Duration::from_millis(100)).await;\n        }\n    });\n\n    // Konsument (reaktiv)\n    while let Some(msg) = receiver.recv().await {\n        println!("Erhalten: {}", msg);\n        if msg == 4 {\n            break;\n        }\n    }\n}\n

\n\n

Kanäle eignen sich ideal für die Implementierung eines Event Bus, einer Ereignisbus oder einfacher reaktiver Systeme mit mehreren Datenquellen.

\n\n

Fortgeschrittene Muster: Kombinieren von Streams

\n