rust-rdkafka

一个完全异步的，由 futures 启用的基于 librdkafka 的 Rust-based Apache Kafka 客户端库。

库

rust-rdkafka 为 librdkafka 提供了安全的 Rust 接口。此版本与 librdkafka v1.9.2+ 兼容。

文档

• 当前 master 分支
• 最新版本
• 变更日志

特性

目前提供的主要功能包括

• 支持 0.8.x 版本以来的所有 Kafka 版本。有关代理兼容性选项的更多信息，请参阅 librdkafka 文档。
• 从单个或多个主题中消费。
• 自动消费者重新平衡。
• 可定制的重新平衡，包括重新平衡前后的回调。
• 同步或异步消息生产。
• 可定制的偏移提交。
• 创建和删除主题以及添加和编辑分区。
• 修改代理和主题配置。
• 访问集群元数据（主题分区列表、副本、活动代理等）。
• 访问组元数据（列出组、列出组成员、主机名等）。
• 访问生产者和消费者指标、错误和回调。
• 通过幂等和事务性生产者和读取提交的消费者实现一次精确语义（EOS）。

每秒一百万条消息

rust-rdkafka 通过在 Rust 中编写的抽象层，设计得易于使用且安全，同时由于 librdkafka C 库而非常快速。

以下是一些使用 BaseProducer 的基准测试结果，向运行在本地的单个 Kafka 0.11 进程发送数据（默认配置，3 个分区）。硬件：戴尔笔记本电脑，搭载英特尔酷睿 i7-4712HQ @ 2.30GHz。

• 场景：生产 500 万条消息，每条消息 10 字节，等待所有消息被确认

  • 1045413 条消息/秒，9.970 MB/秒（5 次运行的平均值）

• 场景：生产 100000 条消息，每条消息 10 KB，等待所有消息被确认

  • 24623 条消息/秒，234.826 MB/秒（5 次运行的平均值）

有关更多数据，请参阅 kafka-benchmark 项目。

客户端类型

rust-rdkafka 提供了低级和高级消费者和生产者。

低级

• BaseConsumer：librdkafka消费者包装的一个简单封装。为了执行回调、重新平衡和接收消息，它必须定期调用poll()。
• BaseProducer：librdkafka生产者的简单封装。与消费者的情况类似，用户必须定期调用poll()以执行交付回调。
• ThreadedProducer：一个具有专门线程用于轮询生产者的BaseProducer。

高级

• StreamConsumer：自动轮询消费者的消息流。
• FutureProducer：一个一旦消息被交付到Kafka（或失败）就会完成的Future。

有关消费者和生产者的更多信息，请参阅它们模块级别的文档。

警告：该库正在积极开发中，API可能会更改。

使用Tokio的异步数据处理

Tokio是一个用于在Rust中快速处理异步事件的平台。由StreamConsumer和FutureProducer公开的接口允许rust-rdkafka用户轻松地在Tokio平台内集成Kafka消费者和生产者，并编写异步消息处理代码。请注意，可以在没有Tokio的情况下使用rust-rdkafka。

要查看rust-rdkafka与Tokio一起工作的示例，请查看示例文件夹中的异步处理示例。

至少一次交付

在许多流应用中，至少一次交付语义很常见：每条消息都保证至少被处理一次；在临时失败的情况下，消息可以被重新处理和/或重新交付，但不会有消息丢失。

为了实现至少一次交付，流处理应用程序必须仔细提交偏移量，只有在消息被处理后才能提交。相反，过早提交偏移量可能会导致消息丢失，因为恢复后消费者将从下一条消息开始，跳过失败发生的消息。

要查看如何使用rdkafka实现至少一次交付，请查看示例文件夹中的至少一次交付示例。要了解更多关于交付语义的信息，请查看Kafka文档中的消息交付语义章节。

精确一次语义

可以使用事务性生产者实现精确一次语义（EOS），这允许原子性地提交或中止生产的记录和消费者偏移量。将isolation.level设置为read_committed的消费者将只会观察到已提交的消息。

EOS在需要消息被精确一次处理的读取-处理-写入场景中非常有用。

要了解如何在 rust-rdkafka 中使用事务，请参阅生产者文档中的 事务 部分。

用户

以下是使用 rust-rdkafka 的部分项目

• timely-dataflow：一个分布式数据并行计算引擎。还可以参见宣布其 Kafka 集成的 博客文章。
• kafka-view：Kafka 集群的 Web 界面。
• kafka-benchmark：Kafka 的高性能基准测试工具。
• callysto：Rust 的流处理框架。
• bytewax：使用 Timely Dataflow 的 Python 流处理框架。

如果您正在使用 rust-rdkafka，请告诉我们！

安装

将此添加到您的 Cargo.toml

[dependencies]
rdkafka = { version = "0.25", features = ["cmake-build"] }

此包将从源编译 librdkafka 并将其静态链接到您的可执行文件。要编译 librdkafka，您需要

• GNU 工具链
• GNU make
• pthreads
• zlib：可选，但默认包含（功能：libz）
• cmake：可选，默认不包含（功能：cmake-build）
• libssl-dev：可选，默认不包含（功能：ssl）
• libsasl2-dev：可选，默认不包含（功能：gssapi）
• libzstd-dev：可选，默认不包含（功能：zstd-pkg-config）

请注意，如果您可以使用 CMake 依赖，则建议通过 cmake-build 功能使用 CMake 构建系统。

默认情况下，将使用一个子模块，其中 librdkafka 源代码被固定到特定的提交，以编译和静态链接库。可以使用 dynamic-linking 功能动态链接 rdkafka 到系统版本的 librdkafka。示例

[dependencies]
rdkafka = { version = "0.25", features = ["dynamic-linking"] }

有关功能的完整列表，请参阅 rdkafka-sys 包的文档。rdkafka-sys 的所有功能都作为 rdkafka 功能重新导出。

最低支持的 Rust 版本 (MSRV)

当前最低支持的 Rust 版本 (MSRV) 为 1.61.0。请注意，提高 MSRV 并不视为破坏性变更。rust-rdkafka 的任何版本都可能提高 MSRV。

异步运行时

StreamConsumer 和 FutureProducer 的某些功能依赖于 Tokio，这对于只想使用低级消费者和生产者的用户来说可能是一个重量级的依赖项。Tokio 集成默认启用，但可以通过关闭默认功能来禁用

[dependencies]
rdkafka = { version = "0.25", default-features = false }

cargo run --example <example_name> -- <example_args>

RUST_LOG="librdkafka=trace,rdkafka::client=debug" cargo test