Arbeit mit Kafka in Rust: Bibliotheken und Beispiele für Anfänger

Arbeit mit Kafka in Rust: Überblick über Bibliotheken und praktische Beispiele

Apache Kafka ist eine verteilte Plattform für die Stream-Verarbeitung von Daten, die häufig zum Aufbau zuverlässiger und skalierbarer Nachrichtensysteme verwendet wird. Die Sprache Rust wird aufgrund ihrer Leistung und Speichersicherheit zunehmend zur bevorzugten Wahl für die Entwicklung hochlastiger Anwendungen, die mit Kafka arbeiten. In diesem Artikel betrachten wir die wichtigsten Bibliotheken für die Arbeit mit Kafka in Rust, deren Installation und geben einfache Codebeispiele.

Beliebte Bibliotheken für Kafka in Rust

Derzeit gibt es zwei Hauptansätze für die Arbeit mit Kafka im Rust-Ökosystem:

• rdkafka — ein Wrapper um die beliebte C-Bibliothek librdkafka. Bietet maximale Leistung und Funktionalität, einschließlich Unterstützung aller modernen Kafka-Funktionen (Transaktionen, Idempotenz, exakte Liefergarantien).
• Kafka-Rust (kafka) — eine reine Implementierung des Kafka-Protokolls in Rust, ohne externe Abhängigkeiten. Einfacher zu erstellen, kann aber in der Funktionalität hinter rdkafka zurückbleiben.

Für die meisten Produktionsprojekte wird die Verwendung von rdkafka empfohlen, da es ausgereifter, leistungsfähiger und von der Community aktiv unterstützt wird.

Installation von rdkafka

Fügen Sie zunächst die Abhängigkeit zu Ihrer Cargo.toml hinzu:

[dependencies]rdkafka = { version = "0.36", features = ["cmake-build"] }tokio = { version = "1", features = ["full"] }  # für asynchrone Arbeitserde = { version = "1", features = ["derive"] }  # für Datenserialisierungserde_json = "1"  # für die Arbeit mit JSON

Die Bibliothek rdkafka erfordert librdkafka auf dem System. Sie können es über Ihren Systempaketmanager installieren oder das Feature cmake-build wie im obigen Beispiel verwenden – in diesem Fall wird die Bibliothek automatisch kompiliert.

Beispiel: Einfacher Producer

Der Producer sendet Nachrichten an ein bestimmtes Kafka-Topic. Hier ist ein einfaches Beispiel für einen asynchronen Producer mit Tokio:

use rdkafka::config::ClientConfig;use rdkafka::producer::{FutureProducer, FutureRecord};use std::time::Duration;

#[tokio::main]async fn main() {    // Producer-Konfiguration erstellen    let producer: FutureProducer = ClientConfig::new()        .set("bootstrap.servers", "localhost:9092")        .set("message.timeout.ms", "5000")        .create()        .expect("Fehler beim Erstellen des Producers");

    // Nachricht an das Topic "my-topic" senden    let delivery_status = producer        .send(            FutureRecord::to("my-topic")                .payload("Hallo, Kafka aus Rust!")                .key("nachrichten-schlüssel"),            Duration::from_secs(0),        )        .await;

    match delivery_status {        Ok(partition_offset) => println!("Nachricht gesendet: {:?}", partition_offset),        Err((e, _)) => eprintln!("Fehler beim Senden: {}", e),    }}

Beispiel: Einfacher Consumer

Der Consumer liest Nachrichten aus einem Topic. Nachfolgend ein Beispiel für einen synchronen Consumer mit manuellem Commit:

use rdkafka::config::ClientConfig;use rdkafka::consumer::{Consumer, StreamConsumer};use rdkafka::message::Message;use rdkafka::util::get_rdkafka_version;

fn main() {    let consumer: StreamConsumer = ClientConfig::new()        .set("group.id", "my-consumer-group")        .set("bootstrap.servers", "localhost:9092")        .set("enable.auto.commit", "false")  // manueller Commit        .set("auto.offset.reset", "earliest")  // von Anfang an lesen        .create()        .expect("Fehler beim Erstellen des Consumers");

    consumer.subscribe(&["my-topic"]).expect("Fehler beim Abonnieren");

    loop {        match consumer.recv() {            Ok(msg) => {                let paylo