自建 MQTT 迁移 IoT 物联网平台实战——实践类

作者：阿里云AIoT

2023-03-01
浙江
本文字数：5952 字
阅读完需：约 20 分钟

自建 MQTT 迁移 IoT 物联网平台实战

前言

随着业务增长，自建 MQTT 集群维护成本越来越高，连接稳定性，消息时延不断遇到上层业务挑战，迁移阿里云 IoT 物联网平台是非常明智的最佳选择。

现有业务背景

企业基于自建 MQTT 集群的数据交互如下图：

image.png

设备端 publish 业务数据到 MQTT 集群，流转到后端业务服务器处理；业务服务器把指令发布到 MQTT 集群，触达到设备端。具体业务数据定义如下：

架构方案

当我们从自己 MQTT 集群迁移到 IoT 物联网平台，我们可以采用以下技术方案，实现低成本的快速迁移，减少设备端和业务系统的改造。

IoT 设备迁移上云三步走：

设备做固件升级，修改接入域名为 IoT 平台 Endpoint
调整现有通信 topic 为 IoT 物联网平台产品自定义 Topic
配置规则引擎，把设备数据流转到服务端订阅 AMQP 消费组，业务服务器实时接收到设备数据

image.png

设备迁移上云实战

创建产品

首先，我们接入物联网平台控制台，左侧栏选择产品，创建新产品：充电宝机柜。如下图：

所属品类：自定义品类
节点类型：直连设备
连网方式：WiFi （这里可以根据实际情况选择）
数据格式：ICA 标准数据格式 JSON（这里也可以选透传/自定义）

image.png

然后，把现有系统业务 Topic 映射到自定义通信的 Topic 类，如下表：

最后，我们接入产品详情，选择 Topic 类列表，点击自定义 Topic，添加 Topic 如下图：

image.png

至此，我们完成了在 IoT 物联网平台的产品创建和通信定义。

注册设备

我们基于产品，创建一个具体设备，获取到设备身份认证信息（三元组），如下图：

image.png

数据流转配置

当我们创建完成产品，注册好设备之后，就需要在 IoT 平台配置数据流转方案了。

服务端订阅 AMQP 消费组

首先，我们要在控制台的服务端订阅，创建我们充电宝业务的 AMQP 消费组，用来消费设备生成的数据。如下图：

image.png

然后，我们进入消费组详情，查看消费组状态。由于我们设备端和服务端都没有连接到平台，数据都是为空。如下图：

image.png

规则引擎

接下来我们要做的是配置规则引擎处理数据，并流转到已创建的消费组上。

我们选择云产品流转，点击创建规则，选择二进制格式（二进制泛指非 JSON 结构数据）。如下图：

image.png

然后，我们编写 SQL，让 IoT 物联网平台处理设备数据，并携带设备信息。如下图：

SELECT deviceName() as deviceName,timestamp() as timestamp ,payload() as payload FROM "/a********E/+/user/data/up"

复制代码

image.png

接下来，我们添加数据转发操作到指定的 AMQP 消费组，如下图：

image.png

完整的规则引擎配置如下图所示：

image.png

最后，我们在规则引擎列表，启动规则。如下图：

image.png

设备开发

在完成了云上控制台的配置工作后，我们要做的就是设备端业务开发。这里我们在 Mac 上用 nodejs 脚本模拟设备业务行为，设备 MQTT 连接，数据上报，指令接收。完整代码如下：

// 引入依赖mqtt库，或自己实现const mqtt = require('aliyun-iot-mqtt');// 设备身份var options = {    productKey: "设备pk",    deviceName: "设备dn",    deviceSecret: "设备ds",    regionId: "cn-shanghai"};
// 1.建立连接const client = mqtt.getAliyunIotMqttClient(options);
// 2.设备接收云端指令数据client.on('message', function(topic, message) {    console.log("topic " + topic)    console.log("message " + message)})// 3. 设备订阅指令的Topicclient.subscribe(`/${options.productKey}/${options.deviceName}/user/cmd/down`)
// 4. 模拟设备 上报数据（原始报文）setInterval(function() {    client.publish(`/${options.productKey}/${options.deviceName}/user/data/up`, getPostData(),{qos:1});
}, 1000);

// 模拟 设备原有报文格式function getPostData() {    let payload = "DE02,"+Math.floor((Math.random() * 20) + 10)    +","+Math.floor((Math.random() * 20) + 10)    +",011101010,am024,1d478f";
    console.log("payload=[ " + payload+" ]")    return JSON.stringify(payload);}

复制代码

至此，我们完成了设备端业务开发。

服务端开发

服务端我们以 Java 为例，演示设备数据接收和控制指令的下发。

业务服务器接收设备数据

参考服务端订阅 AMQP 文档 https://help.aliyun.com/document_detail/143601.html完整代码如下：

package com.aliyun.iot;
import org.apache.commons.codec.binary.Base64;import org.apache.qpid.jms.JmsConnection;import org.apache.qpid.jms.JmsConnectionListener;import org.apache.qpid.jms.message.JmsInboundMessageDispatch;import org.slf4j.Logger;import org.slf4j.LoggerFactory;
import javax.crypto.Mac;import javax.crypto.spec.SecretKeySpec;import javax.jms.*;import javax.naming.Context;import javax.naming.InitialContext;import java.net.URI;import java.util.Hashtable;
public class AMQPClient {
    private final static Logger logger = LoggerFactory.getLogger(AMQPClient.class);
    public static void main(String[] args) throws Exception {        //参数说明，请参见上一篇文档：AMQP客户端接入说明。        String accessKey = "阿里云账号ak";        String accessSecret = "阿里云账号as";        String consumerGroupId = "服务端订阅消费组ID";        long timeStamp = System.currentTimeMillis();        //签名方法：支持hmacmd5，hmacsha1和hmacsha256        String signMethod = "hmacsha1";        //控制台服务端订阅中消费组状态页客户端ID一栏将显示clientId参数。        //建议使用机器UUID、MAC地址、IP等唯一标识等作为clientId。便于您区分识别不同的客户端。        String clientId = "ecs_"+System.currentTimeMillis();
        //UserName组装方法，请参见上一篇文档：AMQP客户端接入说明。        String userName = clientId + "|authMode=aksign"                + ",signMethod=" + signMethod                + ",timestamp=" + timeStamp                + ",authId=" + accessKey                + ",consumerGroupId=" + consumerGroupId                + "|";        //password组装方法，请参见上一篇文档：AMQP客户端接入说明。        String signContent = "authId=" + accessKey + "&timestamp=" + timeStamp;        String password = doSign(signContent,accessSecret, signMethod);        //按照qpid-jms的规范，组装连接URL。        String connectionUrl = "failover:(amqps://替换你的阿里云账号UID.iot-amqp.cn-shanghai.aliyuncs.com:5671?amqp.idleTimeout=80000)"                + "?failover.reconnectDelay=30";
        Hashtable<String, String> hashtable = new Hashtable<>();        hashtable.put("connectionfactory.SBCF",connectionUrl);        hashtable.put("queue.QUEUE", "default");        hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");        Context context = new InitialContext(hashtable);        ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF");        Destination queue = (Destination)context.lookup("QUEUE");        // Create Connection        Connection connection = cf.createConnection(userName, password);        ((JmsConnection) connection).addConnectionListener(myJmsConnectionListener);        // Create Session        // Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手动调用message.acknowledge()        // Session.AUTO_ACKNOWLEDGE: SDK自动ACK（推荐）        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);        connection.start();        // Create Receiver Link        MessageConsumer consumer = session.createConsumer(queue);        consumer.setMessageListener(messageListener);    }
    private static MessageListener messageListener = new MessageListener() {        @Override        public void onMessage(Message message) {            try {                byte[] body = message.getBody(byte[].class);                String content = new String(body);                String topic = message.getStringProperty("topic");                String messageId = message.getStringProperty("messageId");                logger.info("receive message"                        + ", topic = " + topic                        + ", messageId = " + messageId                        + ", content = " + content);                System.out.println();                //如果创建Session选择的是Session.CLIENT_ACKNOWLEDGE，这里需要手动ACK。                //message.acknowledge();                //如果要对收到的消息做耗时的处理，请异步处理，确保这里不要有耗时逻辑。            } catch (Exception e) {                e.printStackTrace();            }        }    };
    private static JmsConnectionListener myJmsConnectionListener = new JmsConnectionListener() {        /**         * 连接成功建立。         */        @Override        public void onConnectionEstablished(URI remoteURI) {            logger.info("onConnectionEstablished, remoteUri:{}", remoteURI);        }
        /**         * 尝试过最大重试次数之后，最终连接失败。         */        @Override        public void onConnectionFailure(Throwable error) {            logger.error("onConnectionFailure, {}", error.getMessage());        }
        /**         * 连接中断。         */        @Override        public void onConnectionInterrupted(URI remoteURI) {            logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI);        }
        /**         * 连接中断后又自动重连上。         */        @Override        public void onConnectionRestored(URI remoteURI) {            logger.info("onConnectionRestored, remoteUri:{}", remoteURI);        }
        @Override        public void onInboundMessage(JmsInboundMessageDispatch envelope) {}
        @Override        public void onSessionClosed(Session session, Throwable cause) {}
        @Override        public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {}
        @Override        public void onProducerClosed(MessageProducer producer, Throwable cause) {}    };
    /**     * password签名计算方法，请参见上一篇文档：AMQP客户端接入说明。     */    private static String doSign(String toSignString, String secret, String signMethod) throws Exception {        SecretKeySpec signingKey = new SecretKeySpec(secret.getBytes(), signMethod);        Mac mac = Mac.getInstance(signMethod);        mac.init(signingKey);        byte[] rawHmac = mac.doFinal(toSignString.getBytes());        return Base64.encodeBase64String(rawHmac);    }}

复制代码

业务服务器下发控制指令

参考云端 API 文档 https://help.aliyun.com/document_detail/69793.html

完整代码如下：

package com.aliyun.iot;
import com.aliyuncs.DefaultAcsClient;import com.aliyuncs.IAcsClient;import com.aliyuncs.exceptions.ClientException;import com.aliyuncs.iot.model.v20170420.PubRequest;import com.aliyuncs.iot.model.v20170420.PubResponse;import com.aliyuncs.profile.DefaultProfile;import com.google.gson.Gson;
public class PubClient {
    public static void main(String[] args) {
        DefaultProfile profile = DefaultProfile.getProfile(                "cn-shanghai",                "替换你的ak",                "替换你的as");        IAcsClient client = new DefaultAcsClient(profile);
        PubRequest request = new PubRequest();        request.setTopicFullName("/替换设备pk/替换设备dn/user/cmd/down");        request.setMessageContent("Q01ELDgyOTIzLGFkMzIyCiA=");//原始报文 : CMD,82923,ad322        request.setProductKey("替换设备pk");        request.setQos(1);
        try {            PubResponse response = client.getAcsResponse(request);            System.out.println(new Gson().toJson(response));
        } catch (ClientException e) {            System.out.println("ErrCode:" + e.getErrCode());            System.out.println("ErrMsg:" + e.getErrMsg());            System.out.println("RequestId:" + e.getRequestId());        }    }}