azure-eventhub-rust

Rust 用 Azure Event Hubs SDK

Azure Event Hubs 向けのクライアントライブラリです。Azure Event Hubs はビッグデータストリーミングプラットフォームおよびイベント取り込みサービスです。

以下の例を使用するには、互換性を確認済みのクレート一式を用意し、すべてのリソースのプレースホルダーを置き換える必要があります。デモ用の書き込みに本番データを使用しないでください。

インストール

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 のベータ系列を使用するため、公開されている対応する identity のベータ版と組み合わせてください。安定版の core 系列のクレデンシャル特性を混在させないでください。

レビュー済みのサンプル一式では、azure_messaging_eventhubs 0.16.0 と互換性のある Azure core／クレデンシャルクレートを使用しています。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 クレートと、必要な Tokio／Serde／Futures 依存関係を含む非同期アプリケーションコード内の断片です。使用前に、選択したバージョンの API を確認してください。公式ソース／リファレンスに照らしてレビューされていますが、ここではコンパイルもクラウドアカウントに対する実行も行っていません。
• 最小権限の原則に従い、認可されたテナント、リソース、ID を使用してください。書き込み、上書き、削除、バックアップの取得、アカウントの変更には、それぞれ対応するスコープが必要です。クリーンアップの対象は、明示的に特定された破棄可能なリソースに限ってください。リクエスト、再試行、ダウンロードするバイト数に上限を設け、ペイロード、クレデンシャル、シークレット値をログに記録しないでください。
• 出力を、環境固有の検証、テスト、専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を停止して確認してください。