Работа с Kafka в Rust: библиотеки rdkafka и Franz

Работа с Kafka в Rust: обзор библиотек

Apache Kafka — это распределённая платформа для потоковой передачи данных, широко используемая в микросервисной архитектуре, обработке событий и системах реального времени. Язык Rust, благодаря своей производительности и безопасности памяти, отлично подходит для написания высоконагруженных Kafka-приложений. В этой статье мы рассмотрим две основные библиотеки для работы с Kafka в Rust: rdkafka (на основе librdkafka) и Franz (чистый Rust).

Почему стоит выбрать Rust для работы с Kafka?

• Высокая производительность — Rust компилируется в нативный код, что минимизирует накладные расходы.
• Безопасность памяти — система владения предотвращает гонки данных и утечки памяти.
• Нулевая стоимость абстракций — позволяет писать эффективный код без потери выразительности.
• Асинхронность — встроенная поддержка async/await упрощает работу с потоками.

Библиотека rdkafka

rdkafka — это самая популярная библиотека для работы с Kafka в Rust. Она представляет собой обёртку над librdkafka — высокопроизводительной C-библиотекой, используемой во многих проектах (включая Node.js, Python, Go).

Установка rdkafka

Добавьте зависимость в Cargo.toml:

[dependencies]
rdkafka = { version = "0.36", features = ["cmake-build"] }
tokio = { version = "1", features = ["full"] }

Пример: Producer (отправка сообщений)

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

#[tokio::main]
async fn main() {
    let producer: FutureProducer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .create()
        .expect("Producer creation error");

    let record = FutureRecord::to("my-topic")
        .payload("Hello from Rust!")
        .key("key-1");

    let delivery_status = producer
        .send(record, Duration::from_secs(0))
        .await;

    match delivery_status {
        Ok(delivery) => println!("Sent: {:?}", delivery),
        Err(e) => eprintln!("Error: {:?}", e),
    }
}

Пример: Consumer (чтение сообщений)

use rdkafka::consumer::{Consumer, StreamConsumer};
use rdkafka::ClientConfig;
use rdkafka::message::Message;
use futures::stream::StreamExt;

#[tokio::main]
async fn main() {
    let consumer: StreamConsumer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("group.id", "my-group")
        .set("auto.offset.reset", "earliest")
        .create()
        .expect("Consumer creation error");

    consumer.subscribe(&["my-topic"]).unwrap();

    let mut stream = consumer.stream();
    while let Some(message) = stream.next().await {
        match message {
            Ok(msg) => {
                let payload = msg.payload_view::<str>().unwrap_or(None);
                if let Some(text) = payload {
                    println!("Received: {}", text);
                }
            }
            Err(e) => eprintln!("Error: {}", e),
        }
    }
}

Библиотека Franz

Franz — это современная библиотека для Kafka, написанная полностью на Rust (без привязок к C). Она предоставляет безопасный и эргономичный API, использует async/await и поддерживает последние версии протокола Kafka (включая KIP-848).

Установка Franz

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

Пример: Producer в Franz

use franz::client::Client;
use franz::record::Record;
use franz::producer::Producer;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::new(vec!["localhost:9092".to_string()]).await?;
    let mut producer = Producer::new(client);

    let record = Record::new("my-topic", Some("key-2"), "Hello from Franz!");
    producer.send(record).await?;
    println!("Message sent");

    Ok(())
}

Пример: Consumer в Franz

use franz::client::Client;
use franz::consumer::{Consumer, ConsumerConfig};
use futures::stream::StreamExt;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::new(vec!["localhost:9092".to_string()]).await?;
    let config = ConsumerConfig::new("my-group", "my-topic");
    let mut consumer = Consumer::new(client, config);

    while let Some(record) = consumer.next().await {
        match record {
            Ok(msg) => println!("Got: {:?}", msg),
            Err(e) => eprintln!("Error: {}", e),
        }