文档

云消息队列 MQTT 版

更新时间:
一键部署

本文介绍如何在云消息队列 RocketMQ 版控制台创建消息流入任务,以实现将云消息队列 MQTT 版的数据流入至云消息队列 RocketMQ 版

前提条件

  • 您已购买云消息队列 RocketMQ 版实例,且实例处于服务中状态。具体步骤,请参见创建资源

  • 您已创建云消息队列 MQTT 版实例,且实例处于服务中状态。具体步骤,请参见创建资源

开服地域

在消息集成中创建消息流入任务时,所有地域都支持将云消息队列 MQTT 版作为消息流入任务源端。

创建消息流入任务

  1. 登录云消息队列 RocketMQ 版控制台,在左侧导航栏选择消息集成 > 任务列表

  2. 在顶部菜单栏选择地域,如华东1(杭州),然后在任务列表页面中,单击创建任务列表

  3. 创建任务面板,设置任务名称描述,配置以下参数。

    • 任务创建

      1. Source(源)配置向导,选择数据提供方微消息队列 MQTT版,设置以下参数,然后单击下一步

        参数

        说明

        示例

        MQTT 实例

        选择MQTT实例。

        test-instance

        MQTT Topic

        选择MQTT实例中的Topic。

        test-topic

        数据格式(Body)

        数据格式是针对支持二进制传递的数据源端推出的指定内容格式的编码能力。支持多种数据格式编码,如无特殊编码诉求可将格式设置为Json。

        • Json(默认Json格式编码,二进制数据按照utf-8 编码为Json格式放入Payload。)

        • Text(文本格式编码,二进制数据按照utf-8编码为字符串放入Payload。)

        • Binary(二进制格式编码,二进制数据按照Base64编码为字符串放入Payload。)

        Json

        批量推送条数

        调用函数发送的最大批量消息条数,当积压的消息数量到达设定值时才会发送请求,取值范围为 [1,10000]。

        100

        批量推送间隔(单位:秒)

        调用函数的间隔时间,系统每到间隔时间点会将消息聚合后发给函数计算,取值范围为[0,15],单位为秒。0秒表示无等待时间,直接投递。

        3

      2. Filtering(过滤)配置向导,定义数据模式过滤发送的请求。更多信息,请参见消息过滤

      3. Transform(转换)配置向导,设置数据清洗,实现分割、映射、富化及动态路由等繁杂数据加工能力。更多信息,请参见使用函数计算实现消息数据清洗

      4. Sink(目标)配置向导,选择服务类型消息队列 RocketMQ 版,配置以下参数。

        参数

        说明

        示例

        版本

        选择云消息队列 RocketMQ 版实例的版本。

        • RocketMQ 4.x:服务端4.x版本。

        • RocketMQ 5.x:服务端5.x版本。

        RocketMQ 5.x

        实例ID

        选择云消息队列 RocketMQ 版消息的目标实例。

        rmq-cn-****

        Topic

        选择目标实例的Topic。

        topic

        消息体(body)

        • 完整数据

        • 数据提取

        • 固定值

        • 模板

        数据提取

        $.data.body

        自定义属性(Properties)

        • 数据提取

        • 模板

        模板

        变量

        {
          "userProperties":"$.data.userProperties",
          "msgId":"$.data.systemProperties.UNIQ_KEY"
        }

        模板

        {
          "EB_SYS_EMBED_OBJECT":"${userProperties}",
          "UNIQ_KEY":"${msgId}"
        }

        消息索引(Keys)

        • 数据提取

        • 固定值。

        • 模板

        数据提取

        $.data.systemProperties.KEYS

        消息标签(Tags)

        • 数据提取

        • 固定值

        • 模板

        数据提取

        $.data.systemProperties.TAGS
    • 任务属性

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

  4. 完成上述配置后,单击保存。在任务列表页面,找到刚创建的任务,此时状态栏为启动中,当状态变为运行中时,任务创建成功。

其他操作

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

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

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

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

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