Работа с Kafka в Rust: библиотеки и примеры для начинающих

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

Apache Kafka — это распределённая платформа для потоковой обработки данных, широко используемая для построения надёжных и масштабируемых систем обмена сообщениями. Язык Rust, благодаря своей производительности и безопасности памяти, становится всё более популярным выбором для написания высоконагруженных приложений, работающих с Kafka. В этой статье мы рассмотрим основные библиотеки для работы с Kafka в Rust, их установку и приведём простые примеры кода.

Популярные библиотеки для Kafka в Rust

На данный момент существует два основных подхода к работе с Kafka в экосистеме Rust:

• rdkafka — обёртка над популярной C-библиотекой librdkafka. Предоставляет максимальную производительность и функциональность, включая поддержку всех современных возможностей Kafka (транзакции, идемпотентность, точные гарантии доставки).
• Kafka-Rust (kafka) — чистая реализация протокола Kafka на Rust, без внешних зависимостей. Легче в сборке, но может отставать по функциональности от rdkafka.

Для большинства production-проектов рекомендуется использовать rdkafka, так как он более зрелый, производительный и активно поддерживается сообществом.

Установка rdkafka

Для начала работы добавьте зависимость в ваш Cargo.toml:

[dependencies]
rdkafka = { version = "0.36", features = ["cmake-build"] }
tokio = { version = "1", features = ["full"] }  # для асинхронной работы
serde = { version = "1", features = ["derive"] }  # для сериализации данных
serde_json = "1"  # для работы с JSON

Библиотека rdkafka требует наличия librdkafka в системе. Вы можете установить её через системный пакетный менеджер или использовать feature cmake-build, как в примере выше — в этом случае библиотека скомпилируется автоматически.

Пример: базовый продюсер (Producer)

Продюсер отправляет сообщения в указанный топик Kafka. Вот простейший пример асинхронного продюсера на Tokio:

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

#[tokio::main]
async fn main() {
    // Создаём конфигурацию продюсера
    let producer: FutureProducer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("message.timeout.ms", "5000")
        .create()
        .expect("Ошибка создания продюсера");

    // Отправляем сообщение в топик "my-topic"
    let delivery_status = producer
        .send(
            FutureRecord::to("my-topic")
                .payload("Привет, Kafka из Rust!")
                .key("ключ-сообщения"),
            Duration::from_secs(0),
        )
        .await;

    match delivery_status {
        Ok(partition_offset) => println!("Сообщение отправлено: {:?}", partition_offset),
        Err((e, _)) => eprintln!("Ошибка отправки: {}", e),
    }
}

Пример: базовый консьюмер (Consumer)

Консьюмер читает сообщения из топика. Ниже приведён пример синхронного консьюмера с ручным подтверждением (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")  // ручной commit
        .set("auto.offset.reset", "earliest")  // читать с начала
        .create()
        .expect("Ошибка создания консьюмера");

    consumer.subscribe(&["my-topic"]).expect("Ошибка подписки");

    loop {
        match consumer.recv() {
            Ok(msg) => {
                let payload = msg.payload().unwrap_or(b"");
                let key = msg.key().unwrap_or(b"");
                println!(
                    "Получено: key={:?}, payload={:?}, topic={}, partition={}, offset={}",
                    String::from_utf8_lossy(key),
                    String::from_utf8_lossy(payload),
                    msg.topic(),
                    msg.partition(),
                    msg.offset()
                );
                // Вручную подтверждаем обработку сообщения
                consumer.commit_message(&msg, rdkafka::consumer::CommitMode::Async).unwrap();
            }
            Err(e) => eprintln!("Ошибка получения: {}", e),
        }
    }
}

Работа с сериализацией данных

Часто требуется отправлять не просто строки, а структурированные данные (например, JSON). Для этого удобно использовать библиотеку serde:

use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct