Function Compute:ApsaraMQ for Kafkaトリガー

概要

Function Computeコンソールでトリガーを作成するリクエストを送信すると、Function Computeはトリガーの設定に基づいてEventBridgeにイベントストリームを自動的に作成します。

トリガーの作成後、Function Computeコンソールでトリガーに関する情報を表示できます。 EventBridgeコンソールで、作成されたリソースに関する情報を表示することもできます。 メッセージがApsaraMQ for Kafkaでキューに入れられると、function Computeの関連する関数が呼び出されます。 1つ以上のメッセージイベントがバッチで関数にプッシュされ、バッチ設定に基づいて処理されます。

使用上の注意

• トリガーソースとして機能するApsaraMQ for Kafkaインスタンスは、Function Computeで呼び出される関数と同じリージョンに存在する必要があります。

• 作成されたイベントストリームの数が上限に達した場合、ApsaraMQ for Kafkaトリガーは作成できません。 イベントストリームの上限については、「制限」をご参照ください。

前提条件

• EventBridge

  • EventBridgeを有効にしてRAMユーザーに権限を付与

• Function Compute

  • イベント関数の作成

• ApsaraMQ for Kafka

  • ApsaraMQ for Kafkaインスタンスが作成されました。 詳細については、次をご参照ください： ステップ2: インスタンスの購入とデプロイ

  • トピックとコンシューマーグループが作成されます。 詳細については、「手順3: リソースの作成」をご参照ください。

手順1: ApsaraMQ for Kafkaトリガーの作成

ApsaraMQ for Kafkaインスタンスを作成した場合は、Function Computeコンソールにログインし、目的の関数をクリックして関数の詳細ページに移動し、[設定] タブをクリックし、[トリガー] ページの [トリガーの作成] をクリックします。 トリガーの作成後、[OK] をクリックします。 詳細は以下の図をご参照ください。

ステップ2: (オプション) 入力パラメーターの設定

ApsaraMQ for Kafkaのイベントソースは、入力パラメーターとしてeventの形式で関数に渡されます。 コードを使用して、イベントを解析および処理できます。 eventを関数に手動で渡して、トリガーイベントをシミュレートできます。

1. 関数の詳細ページの [コード] タブで、[テスト関数] の隣のアイコンをクリックし、ドロップダウンリストから [テストパラメーターの設定] を選択します。

2. [テストパラメーターの設定] パネルで、[新しいテストイベントの作成] または [既存のテストイベントの変更] タブをクリックし、イベント名とイベント内容を入力し、[OK] をクリックします。

   次のサンプルコードは、イベントコンテンツの形式の例を示しています。

   [
       {
           "specversion":"1.0",
           "id":"8e215af8-ca18-4249-8645-f96c1026****",
           "source":"acs:alikafka",
           "type":"alikafka:Topic:Message",
           "subject":"acs:alikafka_pre-cn-i7m2t7t1****:topic:mytopic",
           "datacontenttype":"application/json; charset=utf-8",
           "time":"2022-06-23T02:49:51.589Z",
           "aliyunaccountid":"164901546557****",
           "data":{
               "topic":"****",
               "partition":7,
               "offset":25,
               "timestamp":1655952591589,
               "headers":{
                   "headers":[
   
                   ],
                   "isReadOnly":false
               },
               "key":"keytest",
               "value":"hello kafka msg"
           }
       },
       {
           "specversion":"1.0",
           "id":"8e215af8-ca18-4249-8645-f96c1026****",
           "source":"acs:alikafka",
           "type":"alikafka:Topic:Message",
           "subject":"acs:alikafka_pre-cn-i7m2t7t1****:topic:mytopic",
           "datacontenttype":"application/json; charset=utf-8",
           "time":"2022-06-23T02:49:51.589Z",
           "aliyunaccountid":"164901546557****",
           "data":{
               "topic":"****",
               "partition":7,
               "offset":25,
               "timestamp":1655952591589,
               "headers":{
                   "headers":[
   
                   ],
                   "isReadOnly":false
               },
               "key":"keytest",
               "value":"hello kafka msg"
           }
       }
   ]

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

   次の表に、dataに含まれるパラメーターを示します。

   パラメーター	データ型	例	詳細
   トピック	String	TopicName	トピック名。
   パーティション	Int	1	ApsaraMQ for Kafkaインスタンスのパーティションに関する情報。
   オフセット	Int	0	ApsaraMQ for Kafkaインスタンスのメッセージオフセット。
   timestamp	String	1655952591589	メッセージの消費がいつ開始されたかを示すタイムスタンプ。

ステップ3: 関数コードの書き込みとテスト

トリガーの作成後、関数コードを記述し、関数をテストして、コードが正しいかどうかを確認できます。 ApsaraMQ for Kafkaイベントが発生すると、トリガーによって関数が自動的に呼び出されます。

1. On theコード関数の詳細ページのタブで、コードエディターでコードを記述し、コードのデプロイ.

   この例では、関数コードはNode.jsで記述されています。

   'use strict';
   /*
   To enable the initializer feature
   please implement the initializer function as below:
   exports.initializer = (context, callback) => {
     console.log('initializing');
     callback(null, '');
   };
   */
   exports.handler = (event, context, callback) => {
     console.log("event: %s", event);
     // Parse the event parameters and process the event. 
     callback(null, 'return result');
   }

2. 関数をテストします。

   方法1: eventを設定してイベントソースをシミュレートする場合は、[Test Function] をクリックします。

   方法2: ApsaraMQ for Kafkaコンソールにログインし、作成したトピックを選択して、[メッセージの送信] をクリックします。 以下の図は一例です。

   

3. 実行が完了したら、結果をリアルタイムログで表示します。