使用Kafka触发器监听Kafka消息并自动触发函数执行响应消息事件

消息队列 Kafka 版作为事件源通过事件总线EventBridge与函数计算集成后，通过消息队列 Kafka 版触发器（以下简称Kafka触发器）能够触发关联函数执行，通过函数可以对发布到消息队列 Kafka 版的消息进行自定义处理。本文介绍如何在函数计算控制台创建Kafka触发器、配置入口参数以及编写代码并测试代码。

功能简介

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

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

注意事项

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

前提条件

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

步骤一：创建Kafka触发器

登录函数计算控制台，找到目标函数，在函数详情页面，选择触发器页签，单击创建触发器。

在创建触发器面板，触发器类型选择消息队列Kafka版，设置以下配置项，然后单击确定。

配置项	操作	本文示例
名称	填写自定义的触发器名称。	kafka-trigger
版本或别名	默认值为LATEST，如果您需要创建其他版本或别名的触发器，需先在函数详情页的右上角切换到该版本或别名。关于版本和别名的简介，请参见版本管理和别名管理。	LATEST
Kafka 实例	选择已创建的Kafka实例。	alikafka-serverless-cn-****
Topic	选择已创建的Kafka实例的Topic。	topic1
Group ID	快速创建或选择已有的Group ID。注意不要与已有业务共用同一个Group ID，以免影响已有业务消息收发。	GID_FC_TRIGGER_****
消费位点	选择消息的消费位点，即消息队列 Kafka 版从事件总线开始拉取消息的位置。取值说明如下。最早位点：从最早位点开始消费。最新位点：从最新位点开始消费。	最早位点
调用方式	选择函数调用方式。取值说明如下。同步调用：适用于顺序调用场景。单个（批）事件触发函数调用，等待函数执行完成返回结果后，再由下一个（批）事件继续触发函数调用。同步调用请求正文有效负载最大为32 MB。更多信息，请参见同步调用。异步调用：可以快速消费事件。单个（批）事件触发函数调用，函数计算会立刻返回响应，再由下一个（批）事件继续触发函数调用。该过程中函数会异步执行。异步调用请求正文有效负载最大为128 KB。更多信息，请参见异步调用。	同步调用
投递并发最大值	Kafka消息投递到函数计算的并发最大值，取值范围为1~300。该参数仅对同步调用生效。如果需要更高的并发，请进入EventBridge配额中心申请配额名称为EventStreaming FC Sink 同步投递最大并发数的配额。	1

关于推送配置、重试和死信等高级配置项说明，请参见触发器高级功能。

步骤二：（可选）配置函数入口参数

消息队列 Kafka 版事件源会以event的形式作为输入参数传递给函数，您可以使用代码解析event参数，并对event进行处理。您可以手动将event传给函数模拟触发事件。

在函数详情页面的代码页签，单击测试函数右侧的图标，从下拉列表中，选择配置测试参数。

在配置测试参数面板，选择创建新测试事件或编辑已有测试事件，填写事件名称和事件内容，然后单击确定。

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 版事件发生时，触发器会自动触发函数的执行。

在函数详情页面的代码页签，在代码编辑器中编写代码，然后单击部署代码。

本文以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');
}

测试函数。
方式一：如果您是配置函数入口参数event模拟事件源，单击测试函数。
方式二：登录云消息队列 Kafka 版控制台，找到目标Kafka实例的目标Topic，单击体验发送消息，然后根据界面提示发送消息。
执行完成后，在目标函数详情页面的代码页签，单击实时日志查看结果。