在 Rust 中使用 Kafka：rdkafka 和 Franz 库

在 Rust 中使用 Kafka：库概述

Apache Kafka 是一个分布式流数据平台，广泛应用于微服务架构、事件处理和实时系统。Rust 语言凭借其高性能和内存安全性，非常适合编写高负载的 Kafka 应用程序。在本文中，我们将介绍两个用于在 Rust 中使用 Kafka 的主要库：rdkafka（基于 librdkafka）和 Franz（纯 Rust 实现）。

为什么选择 Rust 来处理 Kafka？

• 高性能 — Rust 编译为原生代码，最大限度地减少了开销。
• 内存安全 — 所有权系统防止数据竞争和内存泄漏。
• 零成本抽象 — 允许编写高效代码而不损失表现力。
• 异步支持 — 内置 async/await 支持简化了流处理。

rdkafka 库

rdkafka 是 Rust 中最流行的 Kafka 库。它是 librdkafka 的封装，librdkafka 是一个高性能的 C 库，被许多项目（包括 Node.js、Python、Go）使用。

安装 rdkafka

在 Cargo.toml 中添加依赖：

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

示例：生产者（发送消息）

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),    }}

示例：消费者（读取消息）

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 是一个完全用 Rust 编写的现代 Kafka 库（没有 C 绑定）。它提供安全且符合人体工程学的 API，使用 async/await，并支持最新版本的 Kafka 协议（包括 KIP-848）。

安装 Franz

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

示例：Franz 中的生产者

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

#[tokio::main]async fn main() -> Result<(), Box<dyn std::error::Error>> {    l