rust-rdkafka

这是一个完全异步的、支持 futures 的 Rust 版 Apache Kafka 客户端库，基于 librdkafka 开发。

关于本库

rust-rdkafka 为 librdkafka 提供了一个安全的 Rust 接口。当前版本兼容 librdkafka v1.9.2+。

文档

• 当前主分支
• 最新发布版本
• 更新日志

特性

目前提供的主要特性包括：

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

每秒百万级消息处理

rust-rdkafka 设计目标是易用且安全，这得益于 Rust 编写的抽象层，同时由于底层使用 librdkafka C 库，性能极其出色。

以下是使用 BaseProducer 进行的一些基准测试结果，数据发送到本地运行的单个 Kafka 0.11 进程（默认配置，3 个分区）。硬件：Dell 笔记本，配备 Intel Core i7-4712HQ @ 2.30GHz。

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

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

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

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

更多数据请查看 kafka-benchmark 项目。

客户端类型

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

低级：

• BaseConsumer：librdkafka 消费者的简单封装。必须定期调用 poll() 以执行回调、再平衡和接收消息。
• BaseProducer：librdkafka 生产者的简单封装。与消费者类似，用户必须定期调用 poll() 以执行投递回调。
• ThreadedProducer：带有专门用于轮询生产者的单独线程的 BaseProducer。

高级：

• StreamConsumer：自动处理消费者轮询的消息 Stream。
• FutureProducer：消息成功投递到 Kafka（或失败）后完成的 Future。

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

警告：本库正在积极开发中，API 可能会发生变化。

使用 Tokio 进行异步数据处理

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

要查看结合 Tokio 使用的 rust-rdkafka 示例，请查看示例文件夹中的异步处理示例。

至少一次交付

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

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

要了解如何使用 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"] }

该 crate 将从源代码编译 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 特性来动态链接系统版本的 librdkafka。例如：

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

有关特性的完整列表，请参阅 rdkafka-sys crate 的文档。所有 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 }

如果你想使用除 Tokio 之外的异步运行时,可以通过提供实现 AsyncRuntime trait 的适配器来与 rust-rdkafka 集成。详情请参见以下示例:

• smol
• async-std

示例

你可以在 examples 文件夹中找到示例。运行示例:

cargo run --example <示例名称> -- <示例参数>

调试

rust-rdkafka 使用 log crate 来处理日志。 可选地,启用 tracing 功能以发出 tracing 事件而不是 log 记录。

在测试和示例中,rust-rdkafka 使用 env_logger crate 来格式化日志。在这些上下文中,可以使用 RUST_LOG 环境变量来启用日志记录,例如:

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

这将把 librdkafka 的日志级别配置为 trace,把 Rust 客户端的 client 模块的日志级别配置为 debug。要实际接收来自 librdkafka 的日志,你还需要在生产者或消费者配置中设置 debug 选项(参见 librdkafka 配置)。

要在你的项目中启用调试,请确保使用 env_logger::init() 或任何与 log 兼容的日志框架的等效方法初始化日志记录器。

rdkafka-sys

请参阅 rdkafka-sys。

贡献者

感谢:

• Thijs Cadier - thijsc

替代方案

• kafka-rust: 纯 Rust 实现的 Kafka 客户端。