Trabajo con Kafka en Rust: bibliotecas y ejemplos para principiantes

Trabajo con Kafka en Rust: descripción general de bibliotecas y ejemplos prácticos

Apache Kafka es una plataforma distribuida para el procesamiento de datos en streaming, ampliamente utilizada para construir sistemas de intercambio de mensajes confiables y escalables. El lenguaje Rust, gracias a su rendimiento y seguridad de memoria, se está convirtiendo en una opción cada vez más popular para escribir aplicaciones de alta carga que trabajan con Kafka. En este artículo, revisaremos las principales bibliotecas para trabajar con Kafka en Rust, su instalación y proporcionaremos ejemplos de código simples.

Bibliotecas populares para Kafka en Rust

Actualmente, existen dos enfoques principales para trabajar con Kafka en el ecosistema de Rust:

• rdkafka — envoltorio sobre la popular biblioteca C librdkafka. Proporciona el máximo rendimiento y funcionalidad, incluyendo soporte para todas las capacidades modernas de Kafka (transacciones, idempotencia, garantías de entrega exactas).
• Kafka-Rust (kafka) — implementación pura del protocolo Kafka en Rust, sin dependencias externas. Más fácil de compilar, pero puede quedarse atrás en funcionalidad respecto a rdkafka.

Para la mayoría de los proyectos en producción, se recomienda usar rdkafka, ya que es más maduro, eficiente y cuenta con soporte activo de la comunidad.

Instalación de rdkafka

Para comenzar, agregue la dependencia a su Cargo.toml:

[dependencies]rdkafka = { version = "0.36", features = ["cmake-build"] }tokio = { version = "1", features = ["full"] }  # para trabajo asíncronoserde = { version = "1", features = ["derive"] }  # para serialización de datosserde_json = "1"  # para trabajar con JSON

La biblioteca rdkafka requiere librdkafka en el sistema. Puede instalarla a través del gestor de paquetes del sistema o usar la característica cmake-build, como en el ejemplo anterior; en este caso, la biblioteca se compilará automáticamente.

Ejemplo: productor básico (Producer)

El productor envía mensajes a un tópico de Kafka específico. Aquí hay un ejemplo simple de un productor asíncrono con Tokio:

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

#[tokio::main]async fn main() {    // Creamos la configuración del productor    let producer: FutureProducer = ClientConfig::new()        .set("bootstrap.servers", "localhost:9092")        .set("message.timeout.ms", "5000")        .create()        .expect("Error al crear el productor");

    // Enviamos un mensaje al tópico "my-topic"    let delivery_status = producer        .send(            FutureRecord::to("my-topic")                .payload("¡Hola, Kafka desde Rust!")                .key("clave-del-mensaje"),            Duration::from_secs(0),        )        .await;

    match delivery_status {        Ok(partition_offset) => println!("Mensaje enviado: {:?}", partition_offset),        Err((e, _)) => eprintln!("Error al enviar: {}", e),    }}

Ejemplo: consumidor básico (Consumer)

El consumidor lee mensajes de un tópico. A continuación se muestra un ejemplo de un consumidor síncrono con confirmación manual (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 manual        .set("auto.offset.reset", "earliest")  // leer desde el principio        .create()        .expect("Error al crear el consumidor");

    consumer.subscribe(&["my-topic"]).expect("Error al suscribirse");

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