物联网平台将设备上报的数据流转至消息队列RocketMQ的Topic中后,RocketMQ再将数据流转到企业服务器。本文介绍数据流转的操作步骤。

前提条件

背景信息

数据流转流程图:

消息流转MQ

方案优势:

  • 设备使用MQTT协议接入物联网平台,数据传输链路支持TLS加密,保障数据不被篡改。MQTT协议说明,请参见MQTT协议规范
  • 通过消息队列RocketMQ削峰填谷,缓冲消息,减轻服务器同时接收大量设备消息的压力。

操作步骤

  1. 登录物联网平台控制台,创建产品和设备。
    1. 在控制台左上方,选择地域目标实例所在地域,然后在实例概览页面,找到对应实例,单击实例名称。
      本示例选择地域华东2(上海)
    2. 在左侧导航栏选择设备管理 > 产品,单击创建产品,配置参数,单击确认
      本示例中,产品名称为MQ_test所属品类为自定义品类节点类型为直连设备,其他参数使用默认值。
    3. 单击查看产品详情,在产品详情页面,单击Topic类列表 > 自定义Topic,然后单击自定义Topic类,定义一个用于设备上报数据的Topic。
      本示例中,定义的Topic类:/{YourProductKey}/${YourDeviceName}/user/data
    4. 在左侧导航栏选择设备管理 > 设备,单击添加设备,为产品MQ_test创建设备。
      本示例中,创建了一个名称为MQdevice的设备。
  2. 在消息队列RocketMQ控制台,创建Topic和消费者。
    1. 登录消息队列RocketMQ版控制台
    2. 在左侧导航栏选择实例列表,单击创建实例,创建一个4.0系列的标准版实例,地域选择华东2(上海)
      具体操作,请参见创建实例
      重要
      • RocketMQ实例所在地域必须与物联网平台实例所在地域保持一致。
      • 仅支持将数据流转到RocketMQ 4.x版本实例的Topic中。
    3. 实例列表页面,单击实例名称。
    4. 实例详情页面,单击创建 Group,配置如下图所示,然后单击确定
      物联网设备消息
    5. 单击创建Topic,消息类型选择普通消息
      物联网设备消息
    6. 创建消息消费者,然后在RocketMQ控制台查看消费者状态,确保消费者处于在线状态,订阅关系一致。
      本文以调用TCP协议的SDK为例,进行收发消息。SDK获取和使用的详细内容,请参见调用TCP协议的SDK收发普通消息

      Java语言的SDK代码示例如下:

      说明
      import com.aliyun.openservices.ons.api.Action;
      import com.aliyun.openservices.ons.api.ConsumeContext;
      import com.aliyun.openservices.ons.api.Consumer;
      import com.aliyun.openservices.ons.api.Message;
      import com.aliyun.openservices.ons.api.MessageListener;
      import com.aliyun.openservices.ons.api.ONSFactory;
      import com.aliyun.openservices.ons.api.PropertyKeyConst;
      import java.util.Properties;
      public class ConsumerTest {
          public static void main(String[] args) {
              Properties properties = new Properties();
              // 您在控制台创建的 Group ID
              properties.put(PropertyKeyConst.GROUP_ID, "XXX");
              // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
              properties.put(PropertyKeyConst.AccessKey, "${AccessKey}");
              // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
              properties.put(PropertyKeyConst.SecretKey, "${SecretKey}");
              // 设置 TCP 接入域名,到控制台的实例基本信息中查看
              properties.put(PropertyKeyConst.NAMESRV_ADDR,
                  "XXX");
              // 集群订阅方式 (默认)
              // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
              // 广播订阅方式
              // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
              Consumer consumer = ONSFactory.createConsumer(properties);
              consumer.subscribe("iot_to_mq", "*", new MessageListener() { //订阅多个 Tag
                  public Action consume(Message message, ConsumeContext context) {
                      System.out.println("Receive: " + message);
                      return Action.CommitMessage;
                  }
              });
              consumer.start();
              System.out.println("Consumer Started");
          }
      }
  3. 返回物联网平台控制台在对应实例下,设置数据流转规则,将设备上报的数据转发至消息队列(RocketMQ)。
    1. 在左侧导航栏选择规则引擎 > 云产品流转
    2. 在云产品流转页面,单击创建规则
      重要 若当前页面显示新版功能,先单击右上角返回旧版,进入旧版功能页面,再单击创建规则
    3. 输入规则名称MQ流转数据格式选择为JSON,单击确认
    4. 单击编写SQL,设置数据处理SQL,如下图所示,然后单击确认
      物联网设备消息
    5. 单击添加操作设置数据转发目的地,如下图所示,然后单击确认
      物联网设备消息
    6. 所有设置完成后,返回至云产品流转页面,单击MQ流转规则对应的启动
      规则启动后,物联网平台会将规则SQL中定义的设备上报消息转发至消息队列(RocketMQ)的Topic中。
  4. 使用Java SDK模拟设备接入物联网平台,并上报消息。
    1. 下载Java SDK Demo,然后解压。
    2. 在IntelliJ IDEA中,导入Demo包中的示例工程JavaLinkKitDemo。
    3. 在文件device_id.json中输入MQdevice的设备证书信息:productKeydeviceNamedeviceSecret
    4. 在文件src\main\java\com.aliyun.alink.devicesdk.demo\MqttSample.java中修改MQTT Topic为设备上报数据的Topic。
      本示例中,使用的Topic是/{YourProductKey}/${YourDeviceName}/user/data
      /**
       * 发布接口示例
       */
      public void publish() {
              MqttPublishRequest request = new MqttPublishRequest();
              // topic 用户根据实际场景填写
              request.topic = "/" + productKey + "/" + deviceName + "/user/data";
              ......
              ......
      }
    5. 在文件src\main\java\com.aliyun.alink.devicesdk.demo\HelloWorld.java中修改MQTT接入域名为您设备的MQTT接入域名。
      本示例如下,查看接入域名方法,请参见查看实例终端节点
      public void init(final DeviceInfoData deviceInfoData) {
              ......
              ......
              /**
               * 设置 Mqtt 初始化参数
               */
              IoTMqttClientConfig config = new IoTMqttClientConfig();
              config.productKey = deviceInfoData.productKey;
              config.deviceName = deviceInfoData.deviceName;
              config.deviceSecret = deviceInfoData.deviceSecret;
              config.channelHost = "iot-06****.mqtt.iothub.aliyuncs.com:1883";
              ......
              ......
      }
    6. 运行src\main\java\com.aliyun.alink.devicesdk.demo\HelloWorld.java文件,启动设备。
    返回物联网平台控制台在对应的实例下,选择监控运维 > 日志服务,查看该设备的日志信息,发现设备数据成功转发至RocketMQ。物联网设备消息
  5. 在RocketMQ控制台查看消息。
    1. 在本地运行订阅消息队列RocketMQ资源的代码。
      物联网设备消息
    2. 返回消息队列RocketMQ控制台的实例详情页面,单击消息查询,基于Topic或者Message ID查询消息。
      您可单击详情查看并下载流转至消息队列RocketMQ中的消息详情。

      消息内容如下:

      {"deviceName":"MQdevice"}