创建SMQ Source Connector

本文介绍如何在云消息队列 Kafka 版控制台创建消息流入任务,将轻量消息队列(原 MNS)中的数据快速导入云消息队列 Kafka 版

前提条件

开服地域

在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秒的延迟时间,您可以在任务列表页面的状态栏查看启动进度。

其他操作

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

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

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

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

  • 删除任务:单击删除,然后在提示对话框,单击确认