azure-eventhub-rust

Azure Event Hubs SDK for Rust

用于 Azure Event Hubs 的客户端库。Azure Event Hubs 是大数据流处理平台和事件引入服务。

以下示例需要经过审核的兼容 crate 集合，并替换所有资源占位符。演示写入时不要使用生产数据。

安装

cargo add azure_messaging_eventhubs@=0.16.0 azure_identity@=1.1.0-beta.1 azure_core@=1.2.0-beta.1

环境变量

export EVENTHUBS_HOST="<namespace>.servicebus.windows.net"
export EVENTHUB_NAME="<eventhub-name>"

核心概念

• 命名空间 — Event Hubs 的容器
• Event Hub — 为并行处理而分区的事件流
• 分区 — 有序的事件序列
• 生产者 — 向 Event Hub 发送事件
• 消费者 — 从分区接收事件

生产者客户端

创建生产者

use azure_identity::DeveloperToolsCredential;
use azure_messaging_eventhubs::ProducerClient;

let credential = DeveloperToolsCredential::new(None)?;
let producer = ProducerClient::builder()
    .open("<namespace>.servicebus.windows.net", "eventhub-name", credential.clone())
    .await?;

发送单个事件

producer.send_event(vec![1, 2, 3, 4], None).await?;

发送批次

let batch = producer.create_batch(None).await?;
if !batch.try_add_event_data(b"event 1".to_vec(), None)? {
    return Err("Event does not fit in the batch".into());
}
if !batch.try_add_event_data(b"event 2".to_vec(), None)? {
    return Err("Send the existing batch and retry the remaining event in a new batch".into());
}

producer.send_batch(batch, None).await?;

消费者客户端

创建消费者

use azure_messaging_eventhubs::ConsumerClient;

let credential = DeveloperToolsCredential::new(None)?;
let consumer = ConsumerClient::builder()
    .open("<namespace>.servicebus.windows.net", "eventhub-name", credential.clone())
    .await?;

接收事件

// Open receiver for specific partition
use futures::StreamExt;
let receiver = consumer.open_receiver_on_partition("0".to_string(), None).await?;
let mut events = receiver.stream_events();
// Bound the receive loop and handle the no-event timeout without logging payloads.
for _ in 0..100 {
    match tokio::time::timeout(std::time::Duration::from_secs(10), events.next()).await {
        Ok(Some(Ok(event))) => { /* process the authorized event and checkpoint safely */ }
        Ok(Some(Err(error))) => return Err(error.into()),
        Ok(None) | Err(_) => break,
    }
}

获取 Event Hub 属性

let properties = consumer.get_eventhub_properties(None).await?;
println!("Partitions: {:?}", properties.partition_ids);

获取分区属性

let partition_props = consumer.get_partition_properties("0", None).await?;
println!("Last sequence number: {}", partition_props.last_enqueued_sequence_number);

最佳实践

1. 复用客户端 — 创建一次，发送多个事件
2. 使用批次 — 比逐条发送更高效
3. 检查批次容量 — 批次已满时，try_add_event_data 会返回 false
4. 并行处理分区 — 每个分区都可以独立消费
5. 使用消费者组 — 隔离不同的消费应用程序
6. 处理检查点 — 对于分布式消费者，使用 azure_messaging_eventhubs_checkpointstore_blob

检查点存储（可选）

适用于需要检查点的分布式消费者：

cargo add azure_messaging_eventhubs_checkpointstore_blob

Event Hubs 0.16 使用 Azure core beta 系列，因此应搭配已发布的匹配 identity beta；不要混用来自稳定版 core 系列的凭据 trait。

经过审核的示例集合使用 azure_messaging_eventhubs 0.16.0 以及兼容的 Azure core/credential crate。请记录并提交 Cargo.lock；源码/API 审核不等于编译或实时服务测试。

参考链接

资源	链接
API 参考	https://docs.rs/azure_messaging_eventhubs
源代码	https://github.com/Azure/azure-sdk-for-rust/tree/main/sdk/eventhubs/azure_messaging_eventhubs
crates.io	https://crates.io/crates/azure_messaging_eventhubs

适用场景

当 Rust 应用程序要发布单个或批量事件，或在指定的消费者组和检查点策略下消费特定 Event Hubs 分区时使用。

限制

• 示例是异步应用程序代码中的片段，需要搭配兼容且固定版本的 Azure crate，以及必需的 Tokio/Serde/Futures 依赖项；使用前请核实所选版本的 API。示例已根据官方源码/参考资料进行审核，但未在此处编译，也未针对云帐户运行。
• 使用经过授权的租户、资源和身份，并遵循最小权限原则。写入、覆盖、删除、检索备份和更改帐户都需要相应的权限范围；清理时只能删除已明确标识的一次性资源。限制请求、重试次数和下载字节数；不要记录负载、凭据或机密值。
• 请勿将输出视为环境专属验证、测试或专家审核的替代品。
• 如果缺少必要输入、权限、安全边界或成功标准，请停止并请求澄清。