在 Rust 中使用 Kafka：面向初学者的库和示例

在 Rust 中使用 Kafka：库概览与实践示例

Apache Kafka 是一个分布式流数据处理平台，广泛用于构建可靠且可扩展的消息交换系统。Rust 语言凭借其高性能和内存安全性，正成为编写与 Kafka 交互的高负载应用程序越来越受欢迎的选择。在本文中，我们将介绍在 Rust 中使用 Kafka 的主要库、它们的安装方法，并提供简单的代码示例。

Rust 中用于 Kafka 的流行库

目前，在 Rust 生态系统中使用 Kafka 主要有两种方法：

• rdkafka — 对流行的 C 库 librdkafka 的封装。提供最高的性能和功能，包括对 Kafka 所有现代特性（事务、幂等性、精确一次交付保证）的支持。
• Kafka-Rust (kafka) — 纯 Rust 实现的 Kafka 协议，无外部依赖。构建更简单，但功能上可能落后于 rdkafka。

对于大多数生产项目，建议使用 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。您可以通过系统包管理器安装它，或者像上面的示例一样使用 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("你好，来自 Rust 的 Kafka！")                .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 paylo