功能簡介

您在函數計算的控制臺提交觸發器創建請求之后，函數計算會根據觸發器的配置信息，自動在事件總線EventBridge側創建事件流資源。

創建完成后，您可以在函數計算控制臺查看觸發器信息，同時也可以在事件總線EventBridge控制臺查看自動創建的資源信息。當源消息隊列 Kafka 版中有消息入隊時，將會觸發函數計算執行，觸發時會根據您的攢批配置將一個或多個消息事件以批的形式推送到函數中進行處理。

注意事項

• 作為觸發源的消息隊列 Kafka 版實例必須和函數計算的函數在相同的地域。

• 創建的事件流數量超過上限后，將無法再創建Kafka觸發器。關于事件流數量的限制，請參見使用限制。

前提條件

• 事件總線EventBridge

  • 開通事件總線EventBridge并授權

• 函數計算

  • 創建事件函數

• 消息隊列 Kafka 版

  • 創建實例

  • 創建Topic和Group ID

步驟一：創建Kafka觸發器

當您已經創建好Kafka的實例，您需要登錄函數計算控制臺，進入目標函數，選擇配置頁簽，創建觸發器，創建完成點擊確定如下圖所示。

步驟二：（可選）配置函數入口參數

消息隊列 Kafka 版事件源會以event的形式作為輸入參數傳遞給函數，您可以使用代碼解析event參數，并對event進行處理。您可以手動將event傳給函數模擬觸發事件。

1. 在函數詳情頁面的代碼頁簽，單擊測試函數右側的圖標，從下拉列表中，選擇配置測試參數。

2. 在配置測試參數面板，選擇創建新測試事件或編輯已有測試事件，填寫事件名稱和事件內容，然后單擊確定。

   event格式如下所示：

   [
       {
           "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字段包含的參數解釋如下表所示。

   參數	類型	示例值	描述
   topic	String	TopicName	Topic的名稱。
   partition	Int	1	云消息隊列 Kafka 版的消費分區信息。
   offset	Int	0	云消息隊列 Kafka 版的消息位點。
   timestamp	String	1655952591589	開始消費時間戳。

步驟三：編寫函數代碼并測試

觸發器創建完成后，您可以開始編寫并測試函數代碼，以驗證代碼的正確性。在實際操作過程中，當消息隊列 Kafka 版事件發生時，觸發器會自動觸發函數的執行。

1. 在函數詳情頁面的代碼頁簽，在代碼編輯器中編寫代碼，然后單擊部署代碼。

   本文以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);
     // 解析event參數，對event進行處理。
     callback(null, 'return result');
   }

2. 測試函數。

   方式一：如果您是配置函數入口參數event模擬事件源，單擊測試函數。

   方式二：登錄云消息隊列 Kafka 版控制臺，選擇您創建的目標Topic，點擊發送消息，如下圖。

   

3. 執行完成后，在實時日志查看結果。