关于消息中间件:自建MQTT迁移IoT物联网平台实战实践类

2次阅读

共计 8361 个字符，预计需要花费 21 分钟才能阅读完成。

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

前言

随着业务增长，自建 MQTT 集群保护老本越来越高，连贯稳定性，音讯时延一直遇到下层业务挑战，迁徙阿里云 IoT 物联网平台是十分理智的最佳抉择。

现有业务背景

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

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

架构计划

当咱们从本人 MQTT 集群迁徙到 IoT 物联网平台，咱们能够采纳以下技术计划，实现低成本的疾速迁徙，缩小设施端和业务零碎的革新。

IoT 设施迁徙上云三步走：

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

设施迁徙上云实战

创立产品

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

  • 所属品类：自定义品类
  • 节点类型：直连设施
  • 连网形式：WiFi（这里能够依据理论状况抉择）
  • 数据格式：ICA 规范数据格式 JSON（这里也能够选 透传 / 自定义）

    而后，把现有零碎业务 Topic 映射到自定义通信的 Topic 类，如下表：

最初，咱们接入 产品详情 ，抉择Topic 类列表，点击 自定义 Topic，增加 Topic 如下图：

至此，咱们实现了在 IoT 物联网平台的产品创立和通信定义。

注册设施
咱们基于产品，创立一个具体设施，获取到设施身份认证信息（三元组），如下图：

数据流转配置
当咱们创立实现产品，注册好设施之后，就须要在 IoT 平台配置数据流转计划了。

服务端订阅 AMQP 生产组
首先，咱们要在控制台的 服务端订阅，创立咱们充电宝业务的 AMQP 生产组，用来生产设施生成的数据。如下图：

而后，咱们进入生产组详情，查看生产组状态。因为咱们设施端和服务端都没有连贯到平台，数据都是为空。如下图：

规定引擎
接下来咱们要做的是配置规定引擎解决数据，并流转到已创立的生产组上。

咱们抉择 云产品流转 ，点击创立规定，抉择 二进制 格局（二进制泛指非 JSON 构造数据）。如下图：

而后，咱们编写 SQL，让 IoT 物联网平台解决设施数据，并携带设施信息。如下图：

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

接下来，咱们增加数据转发操作到指定的 AMQP 生产组，如下图：

残缺的规定引擎配置如下图所示：

最初，咱们在规定引擎列表，启动 规定。如下图：

设施开发
在实现了云上控制台的配置工作后，咱们要做的就是设施端业务开发。这里咱们在 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. 设施订阅指令的 Topic
client.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());
        }
    }
}

至此，咱们实现了服务端业务开发。

整体运行
在实现以上业务开发之后，咱们先启动服务端程序和 IoT 物联网平台建设连贯，而后启动设施端模仿脚本。

设施上报数据
设施端日志：在设施端日志，咱们看到打印进去的业务报文如下：

服务端日志 ：
在设施打印进去日志同时，咱们能看到服务端实时在打印从 IoT 物联网平台获取到的音讯数据，如下图：

此时，咱们在物联网平台控制台，生产组详情里，也能看到音讯解决速率，沉积量，最初一条音讯解决工夫，以及服务端 ecs 的列表。如下图：

日志服务
在阿里云 IoT 的控制台日志服务里，咱们能够查看设施行为日志：

在日志服务的上行音讯剖析里，能够查看音讯详情，包含 Topic 和 Payload。

也能够跟踪上行音讯的流转过程，如下图：

服务端下发控制指令
咱们在服务端执行 Pub 接口调用，像设施下发控制指令。如下图：

此时，设施端会实时打印进去指令信息，如下图：

同样，咱们能够在控制台的日志服务，上行音讯剖析中，查看到指令音讯流转残缺过程，如下图：

至此，咱们在尽量少改变的前提下，实现了存量设施上云的迁徙工作。

官网流动
[​​1、挑战设施接入量，超低折扣 + 百万代金券等你拿​​
](https://developer.aliyun.com/article/739438?spm=a2c6h.1287363…)
[​​2、物联网平台企业版（包年包月），实例独享，数据隔离，万台设施接入低至 9.5 元 / 天​​
](https://common-buy.aliyun.com/?spm=a2c6h.12873639.article-det…)
[​​3、物联网平台公共实例，小规格套餐，每年 18 元，开发调试 / 小型我的项目交付必备​​
](https://common-buy.aliyun.com/?spm=a2c6h.12873639.article-det…,%22purchase_num%22:1,%22ord_time%22:%221:Year%22%7D)
[​​4、物联网平台公共实例，按用量购买，低至 6 折，过大年流动正在进行中​​
](https://common-buy.aliyun.com/?spm=a2c6h.12873639.article-det…)
​​5、更多服务​​

物联网平台产品介绍详情：​​https://www.aliyun.com/product/iot/iot_instc_public_cn​​

            阿里云物联网平台客户交换群
正文完
消息中间件
发表至： 消息中间件
2023-03-01
 0
关于消息中间件:EMR集群安全认证和授权管理
关于消息中间件:限时领奖消息队列MNS训练营重磅来袭边学习充电边领充电宝～
关于消息中间件:Serverless-极致弹性解构在线游戏行业痛点
关于消息中间件:极客时间大师课深度剖析-RocketMQ50上线啦欢迎免费领取

站内搜索

-「」