创建SMQ Source Connector

开服地域

在Connector生态集成中创建消息流入任务时，支持将轻量消息队列（原 MNS）作为消息流入任务源端的地域有华东1（杭州）、华东2（上海）、华北1（青岛）、华北2（北京）、华北3（张家口）、华北5（呼和浩特）、华南1（深圳）、西南1（成都）、中国香港、美国（硅谷）、美国（弗吉尼亚）、德国（法兰克福）、新加坡、日本（东京）、马来西亚（吉隆坡）和印度尼西亚（雅加达）。

创建消息流入任务

1. 登录云消息队列 Kafka 版控制台，在概览页面的资源分布区域，选择地域。

2. 在左侧导航栏，选择Connector生态集成 > 任务列表。

3. 在任务列表页面，单击创建任务。

4. 在创建任务面板，设置任务名称和描述，配置以下参数，单击保存。

   • 任务创建

     1. 在Source（源）配置向导，选择数据提供方为轻量消息队列（原 MNS），设置以下参数，然后单击下一步。

        参数	说明	示例
        队列名称	选择轻量消息队列（原 MNS）的队列。	test-queue
        Base64 解码	勾选此项会帮助您将轻量消息队列（原 MNS）的数据解码后进行投递。	test
        批量推送	批量推送可帮您批量聚合多个事件，当批量推送条数和批量推送间隔（单位：秒）两者条件达到其一时即会触发批量推送。 例如：您设置的推送条数为100 条，间隔时间为15 s，在10 s内消息条数已达到100条，那么该次推送则不会等15 s后再推送。	开启
        批量推送条数	调用函数发送的最大批量消息条数，当积压的消息数量到达设定值时才会发送请求，取值范围为 [1,10000]。	100
        批量推送间隔（单位：秒）	调用函数的间隔时间，系统每到间隔时间点会将消息聚合后发给函数计算，取值范围为[0,15]，单位为秒。0秒表示无等待时间，直接投递。	3

     2. 在Filtering（过滤）配置向导，设置数据模式内容过滤发送的请求。更多信息，请参见事件模式。

     3. 在Transform（转换）配置向导，设置数据清洗，实现分割、映射、富化及动态路由等繁杂数据加工能力。更多信息，请参见数据清洗。

     4. 在Sink（目标）配置向导，选择服务类型为消息队列 Kafka 版，配置以下参数。

        参数	说明	示例
        实例ID	选择已创建的云消息队列 Kafka 版实例。	test
        Topic	选择已创建实例中的Topic。	test
        确认模式（ack）	选择云消息队列 Kafka 版接收到数据后给客户端发出的确认信号。	None
        消息体（body）	事件总线EventBridge通过JSONPath提取事件中的数据，将指定的事件内容路由到事件目标。	部分事件 $.data.value
        消息键值（Key）	事件总线EventBridge通过JSONPath提取事件中的数据，将指定的事件内容路由到事件目标。	部分事件 $.data.key

   • 任务属性

     配置事件推送失败时的重试策略及错误发生时的处理方式。更多信息，请参见重试和死信。

5. 返回任务列表页面，找到创建好的任务，在其右侧操作列，单击启用。

6. 在提示对话框，阅读提示信息，然后单击确认。

   启用任务后，会有30秒~60秒的延迟时间，您可以在任务列表页面的状态栏查看启动进度。

其他操作

在任务列表页面，找到目标任务，在其右侧操作列，执行其他操作。

• 查看任务详情：单击详情，在任务页面，查看任务的基础信息、任务属性及监控指标。

• 编辑任务配置：单击编辑，在编辑任务面板，修改任务详情及属性。

• 启停任务：单击启用或者停用，然后在提示对话框，单击确认。

• 删除任务：单击删除，然后在提示对话框，单击确认。