ApsaraMQ for MQTT をイベントソースとしたイベントストリームの作成 - EventBridge

IoT デバイスまたはメッセージングクライアントが MQTT トピックに公開する場合、それらのメッセージを処理、ストレージ、または分析のためにダウンストリームサービスにルーティングする必要があることがよくあります。EventBridge でイベントストリームを作成し、ApsaraMQ for MQTT ソースからメッセージを消費してターゲットに配信できます。

ApsaraMQ for MQTT からのイベントストリームは、次のパイプラインに従います。

ソース (ApsaraMQ for MQTT) --> フィルター (オプション) --> 変換 (オプション) --> ターゲット (シンク)

各ステージは次のように機能します。

ステージ	目的
ソース	ご利用の ApsaraMQ for MQTT インスタンス上の MQTT トピックからメッセージを消費します。
フィルター	各イベントをパターンと照合して評価し、一致しないイベントを破棄します。
変換	配信前にイベントデータを再形成します (例: フィールドの抽出やペイロードの再フォーマット)。
ターゲット	処理されたイベントをダウンストリームサービスに配信します。

前提条件

開始する前に、以下を確認してください。

必要な権限が Resource Access Management (RAM) ユーザーに付与された、アクティブ化された EventBridge サービス。詳細については、「EventBridge をアクティブ化し、RAM ユーザーに権限を付与する」をご参照ください。
[実行中] ステータスの ApsaraMQ for MQTT インスタンス。詳細については、「リソースの作成」をご参照ください。

イベントストリームの作成

EventBridge コンソールにログインします。EventBridge コンソールの左側のナビゲーションウィンドウで、[イベントストリーム] をクリックします。
上部ナビゲーションバーでリージョンを選択し、[イベントストリームの作成] をクリックします。
「[イベントストリームの作成]」ページで、「[タスク名]」と「[説明]」を入力し、その後、以下のセクションを完了します。

MQTT ソースの設定

［ソース］ステップで、［データプロバイダー］を［MQTT 用メッセージキュー］に設定します。

ソースパラメーターを設定してから、[次へ] をクリックします。

［バッチプッシュ］が有効になっており、［メッセージ数］を 100、［バッチプッシュ間隔 (単位: 秒)］を 15 に設定した場合、経過時間が 10 秒しか経過していなくても、メッセージ数が 100 に達するとメッセージが送信されます。

パラメーター	説明	例
MQTT インスタンス	メッセージが生成される ApsaraMQ for MQTT インスタンスです。	test-instance
MQTT トピック	ApsaraMQ for MQTT インスタンスでメッセージが生成されるトピックです。	test-topic
バッチプッシュ	一度に複数のイベントを集約します。この機能は、メッセージ数パラメーターまたはバッチプッシュ間隔 (単位: 秒) パラメーターで指定された条件が満たされた場合にトリガーされます。	有効
メッセージ数	各関数呼び出しで送信できるメッセージの最大数です。バックログ内のメッセージ数が指定された値に達した場合にのみ、リクエストが送信されます。有効な値： 1～10000。	100
バッチプッシュ間隔 (単位: 秒)	関数が呼び出される時間間隔です。システムは、指定された時間間隔で集約されたメッセージを Function Compute に送信します。有効な値： 0～15。 `0` は、集約後すぐにメッセージが送信されることを意味します。	3

フィルタリング、変換、およびターゲットの設定

[フィルタリング]、[変換]、および[シンク] の各ステップで、イベントのフィルタリング方法、変換ルール、およびイベントターゲットを設定します。Function Compute を使用したメッセージの変換の詳細については、「Function Compute を使用したメッセージのクリーニング」をご参照ください。

リトライとデッドレターキューの設定

「タスクプロパティ」ステップで、リトライポリシーとデッドレターキューを設定します。ターゲットへの配信に失敗した場合、EventBridge は設定されたポリシーに基づいてリトライを実行します。すべてのリトライ後に依然として配信できないイベントは、検査および再処理のためにデッドレターキューにルーティングされます。リトライオプションの完全な一覧については、「リトライポリシーおよびデッドレターキュー」をご参照ください。

イベントストリームの保存と有効化

[保存] をクリックします。
[イベントストリーム] ページに戻り、該当のイベントストリームを見つけ、[操作] 列の [有効化] をクリックします。有効化には 30～60 秒かかります。[ステータス] 列で進捗状況を確認します。

イベント例

メッセージが MQTT トピックに公開されると、EventBridge はそれを CloudEvents 準拠のエンベロープにラップし、ターゲットに配信します。

{
    "specversion": "1.0",
    "id": "AC1EC0C950650816F27D46F7D7CA****",
    "source": "acs:mqtt",
    "type": "mqtt:Topic:SendMessage",
    "subject": "acs:mq:cn-hangzhou:143998900779****:topic/mqtt-cn-2r42qam****/housekee****",
    "datacontenttype": "application/json; charset=utf-8",
    "time": "2022-06-22T03:53:47.959Z",
    "aliyunaccountid": "143998900779****",
    "data": {
        "props": {
            "firstTopic": "housekee****",
            "secondTopic": "/testMq4****",
            "clientId": "GID_****"
        },
        "body": "TEST"
    }
}

CloudEvents エンベロープフィールド

CloudEvents 仕様で定義されているパラメーターの詳細については、「イベント概要」をご参照ください。

MQTT メッセージペイロード (`data`)

data オブジェクトには、MQTT メッセージペイロードが含まれています。

パラメーター	タイプ	例	説明
`props`	Map	--	メッセージ属性
`props.firstTopic`	String	`housekee****`	メッセージの送受信に使用される親トピック
`props.secondTopic`	String	`/testMq4****`	サブトピック
`props.clientId`	String	`GID_****`	クライアント ID
`body`	String	`TEST`	メッセージ本文