Kafka触发器

功能简介

您在函数计算的控制台提交触发器创建请求之后，函数计算会根据触发器的配置信息，自动在事件总线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. 执行完成后，在实时日志查看结果。