Kafka触发器

功能简介

您在函数计算的控制台提交触发器创建请求之后，函数计算会根据触发器的配置信息，自动在事件总线EventBridge侧创建事件流资源。

创建完成后，您可以在函数计算控制台查看触发器信息，同时也可以在事件总线EventBridge控制台查看自动创建的资源信息。当源消息队列Kafka版中有消息入队时，将会触发函数计算执行，触发时会根据您的攒批配置将一个或多个消息事件以批的形式推送到函数中进行处理。

注意事项

• 作为触发源的消息队列Kafka版实例必须和函数计算的函数在相同的地域。
• 创建的事件流数量超过上限后，将无法再创建Kafka触发器。

前提条件

• 事件总线EventBridge
  • 开通事件总线EventBridge并授权
• 函数计算
  • 创建服务
  • 创建函数
• 消息队列Kafka版
  • 创建实例
  • 创建Topic和Group ID

步骤一：创建Kafka触发器

1. 登录函数计算控制台，在左侧导航栏，单击服务及函数。
2. 在顶部菜单栏，选择地域，然后在服务列表页面，单击目标服务。
3. 在函数管理页面，单击目标函数名称。
4. 在函数详情页面，单击触发器管理页签，从版本或别名下拉列表选择要创建触发器的版本或别名，然后单击创建触发器。
5. 在创建触发器面板，填写相关信息。然后单击确定。
   高级配置的更多信息，请参见EventBridge触发器高级功能。

   配置项类型	配置项	操作	本文示例
   基础配置	触发器类型	选择消息队列 Kafka 版。	消息队列 Kafka 版
   	名称	填写自定义的触发器名称。	kafka-trigger
   	版本或别名	默认值为LATEST，如果您需要创建其他版本或别名的触发器，首先需要在函数详情页的右上角切换到该版本或别名。关于版本和别名的简介，请参见管理版本和管理别名。	LATEST
   	Kafka 实例	选择已创建的消息队列Kafka版实例。	alikafka_pre-cn-i7m2t7t1****
   	Topic	选择已创建的消息队列Kafka版实例的Topic。	topic1
   	Group ID	选择已创建的消息队列Kafka版实例的Group ID。 说明 请使用独立的Group ID来创建触发器，不要与已有的业务混用Group ID，否则会影响已有的消息收发。	GID_group1
   	消费任务并发数	消费者的并发数量，取值范围为[1,Topic的分区数]。	2
   	消费位点	选择消息的消费位点，即消息队列Kafka版从事件总线开始拉取消息的位置。 取值说明如下。 最早位点：从最早位点开始消费。 最新位点：从最新位点开始消费。	最新位点
   	网络配置	选择路由消息的网络类型。 取值说明如下。 默认网络：默认使用部署Kafka实例时选择的VPC ID和vSwitch ID。 自建公网：需选择另外的专有网络VPC、交换机和安全组。	默认网络
   高级配置	调用方式	选择函数调用方式。 取值说明如下。 同步调用：默认调用方式，事件触发函数执行，等待函数调用完成后，函数计算返回执行结果。更多信息，请参见同步调用。 异步调用：适用于调度延时较长的函数，事件触发函数执行后，函数计算立即返回响应结果并且确保函数至少被成功执行一次，但不会返回具体执行结果。更多信息，请参见功能概览。	同步调用
   	批量推送	开启后，可帮您批量聚合多个事件，当批量推送条数和批量推送间隔两者条件达到其一时即会触发批量推送。 例如，您设置的推送条数为100条，间隔时间为15秒，在10秒内消息条数已达到100条，那么该次推送则不会等15秒后再推送。	开启
   	批量推送条数	一次调用函数发送的最大批量消息条数，当积压的消息数量到达设定值时才会发送请求，取值范围为[1,500]。	1
   	批量推送间隔	调用函数的间隔时间，系统每到间隔时间点会将消息聚合后发给函数计算，取值范围为[0,15]，单位为秒。0秒表示无等待时间，直接投递。	1
   	重试策略	消息推送失败时的重试策略。 取值说明如下。 退避重试：重试3次，每次重试的间隔时间是10秒到20秒之间的随机值。 指数衰减重试：重试176次，每次重试的间隔时间指数递增至512秒，总计重试时间为1天；每次重试的具体间隔为：1、2、4、8、...512秒。	退避重试
   	容错策略	当错误发生时的处理方式。 取值说明如下。 允许容错：允许异常容错，当异常发生时不会阻塞执行，超过重试策略后会根据配置将消息投递至死信队列或直接丢弃。 禁止容错：不允许容错，当异常发生并超过重试策略配置时会阻塞执行。	允许容错
   	死信队列	开启后，未处理或超过重试次数的消息可以发送到指定队列。该配置不启用时，超过重试策略后消息将被丢弃。	开启
   	队列类型	死信队列的类型。 取值如下： 消息服务 MNS 消息队列 RocketMQ 版	消息服务 MNS
   	队列名称	死信队列的名称。	test-queue

   创建完成后，在触发器名称列表中显示已创建的触发器。如需对创建的触发器进行修改或删除，具体操作，请参见触发器管理。

步骤二：配置函数入口参数

消息队列Kafka版事件源会以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. 单击函数代码页签的测试函数。
   执行完成后，您可以在函数代码页签的上方查看执行结果。

更多信息

如需对创建的触发器进行修改或删除，具体操作，请参见触发器管理。