智能手持测温枪接入阿里云 IoT 物联网平台实践——实践类

2023-02-28
浙江
本文字数：4092 字
阅读完需：约 13 分钟

1.概述

随着新型冠状病毒疫情发展，社区居家隔离成为有效手段，而体温排查是社区工作的重中之重！借助 IoT 物联网技术可以方便的完成居民体温实时监控和历史数据的完整追溯。

2.技术架构方案

基于稳定性，高并发，低时延的考量我们选择阿里云 IoT 物联网平台搭建整套系统。首先手持测温枪通过蓝牙连接到 DTU 模块，DTU 模块以 MQTT 协议接入物联网平台。数据上云后，通过规则引擎流转服务端订阅的 AMQP 消费组，实时推送到我们业务服务器。管理人员使用手机小程序即可实时看到出入人员的体温数据。

image.png

3.云端开发

3.1 产品创建

进入物联网平台控制台，创建产品。

image.png

在产品详情 Topic 列表，增加用于数据传输的 Topic，如下：

image.png

3.2 注册设备

产品定义好后，我们基于这个产品创建一个具体设备，获取到设备身份三元组。

image.png

3.3 创建消费组

接下来，我们要在服务端订阅创建用来接收数据的消费组，查看下图：

image.png

3.4 配置规则引擎

最后，我们通过规则引，把设备上报的数据做业务处理后，流转到我们服务器的消费组，从而实现企业自己的设备采集的业务数据到达企业自己的后台服务器的流转过程。

image.png

4.设备开发

在完成了云上控制台的配置工作后，我们要做的就是设备端业务开发。这里我们在 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. 模拟设备 上报数据（原始报文）setInterval(function() {    client.publish(`/${options.productKey}/${options.deviceName}/user/data`, getPostData(),{qos:1});
}, 1000);

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

复制代码

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

5.1 业务服务器接收 IoT 数据

参考服务端订阅 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);    //消费组配置参数    private static String accessKey = "阿里云账号ak";    private static String accessSecret = "阿里云账号as";    private static String consumerGroupId = "服务端订阅消费组ID";    private static String aliUID = "替换你的阿里云账号UID";
    public static void main(String[] args) throws Exception {                long timeStamp = System.currentTimeMillis();        //签名方法        String signMethod = "hmacsha1";        //控制台服务端订阅中消费组状态页客户端ID一栏将显示clientId参数。        //建议使用机器UUID、MAC地址、IP等唯一标识等作为clientId。便于您区分识别不同的客户端。        String clientId = "ecs_"+System.currentTimeMillis();
        //UserName组装        String userName = clientId + "|authMode=aksign"                + ",signMethod=" + signMethod                + ",timestamp=" + timeStamp                + ",authId=" + accessKey                + ",consumerGroupId=" + consumerGroupId                + "|";        //password组装        String signContent = "authId=" + accessKey + "&timestamp=" + timeStamp;        String password = doSign(signContent,accessSecret, signMethod);        //按照qpid-jms的规范，组装连接URL。        String connectionUrl = "failover:(amqps://"+aliUID+".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");        // 创建和IoT平台的AMQP连接        Connection connection = cf.createConnection(userName, password);        ((JmsConnection) connection).addConnectionListener(myJmsConnectionListener);        // 创建 Session        // Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手动调用message.acknowledge()        // Session.AUTO_ACKNOWLEDGE: SDK自动ACK（推荐）        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);        connection.start();        // 创建消费者        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);    }}