后台开发之设备通信_iot设备如何和后端通信

后台开发之设备通信

最近接手了一个智能零售柜的项目，负责JAVA后台开发，在熟悉项目过程学习到了很多东西，从项目的框架设计也感受到了框架设计的艺术，比如拦截器、监听器、测试Mock模拟设备（mqtt协议）…还有许许多多可以吸收理解的，比如通过Mybatis plus实现自动生成文件功能、通过Shell脚本来提高开发效率…
今天先记录下关于后台与设备通信之间的一些要注意的点。

MQTT协议

要实现后台与硬件设备的通信，我们采用了MQTT协议来进行通信。
mqtt介绍

MQTT(消息队列遥测传输)是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它工作在 TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议，为此，它需要一个消息中间件 。
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

基本的后台编程逻辑
就是通过设置一个队列（这里采用了阿里云），当需要进行通信的时候，设备也可以发布消息到队列中，由后台服务器订阅消息进行处理（这里是设备会发送自己的一些信息封装为topic，包含设备类型、设备名称、productKey），后台启动线程监听，当发现队列有消息时就进行订阅处理；处理完成后，会发送对应的处理报文返回给设备，设备根据报文执行对应的操作。这里使用的就是订阅者模式，这里要注意的时设备与后台时通过topic这个机制来实现消息的识别的，并且需要设计好设备与服务器的通信协议来识别消息的信息内容。（关于消息报文的一些定义，我的另一篇博客）
这里说说服务器后台的基本实现机制：

订阅：通过设计一个监听器，当程序启动时就进行监听，监听云队列中是否有对应的消息可以订阅，然后通过消息内容进行对应的处理，这里监听的是com.aliyun.mns.client中的CloudQueue。
代码（展示大概逻辑框架）：

@Service("iotMessageQueueListener")
public class IotMessageQueueListener {

    //region 自动注入
    private static Logger logger = LoggerFactory.getLogger(IotMessageQueueListener.class);

    private void log(String s) {
        if (ValueHelper.isNone(s)) {
            return;
        }
        String [] arr = s.split("\n");
        for (String st :arr) {
            logger.info("[" + profile + "]" + st);
        }
    }

    @PostConstruct
    public void start()  {
        threadPoolTaskExecutor.execute(() -> {
            if (SystemTypeEnum.Linux != SystemTypeEnum.getSystem()) {
                logger.info("只能在Linux服务器中执行监听");
                return;
            }

            while (!ApplicationContextListener.isStrartUp) {
                try {
                    logger.info("Spring还在加载中，等候一秒后再试。");
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            profile = SpringContextUtil.getActiveProfile();
            MDC.put("profile", profile);
            MDC.put("traceId", "IOT");

            CloudAccount account = new CloudAccount(
                    aliyunProperties.getAliyunAccessKeyId(),
                    aliyunProperties.getAliyunAccessKeySecret(),
                    aliyunProperties.getMnsEndpoint());

            log("aliyunProperties.MnsEndpoint=" + aliyunProperties.getMnsEndpoint());
            log("aliyunProperties.IotProductKey=" + aliyunProperties.getIotProductKey());

            client = account.getMNSClient();
            CloudQueue queue = client.getQueueRef("aliyun-iot-" + aliyunProperties.getIotProductKey()); //参数请输入IoT自动创建的队列名称，例如上面截图中的aliyun-iot-3AbL0062osF
            int cnt = 0;
            while (true) {
                try {
                    cnt++;
                    Message popMsg = queue.popMessage(10); //长轮询等待时间为30秒
                    threadPoolTaskExecutor.execute(() -> {
                        if (popMsg != null) {
                            queue.deleteMessage(popMsg.getReceiptHandle()); //避免堆积太多消息

                            try {
                                handleMessage(popMsg);//消息处理
                            } catch(Exception ee) {
                                logger.error("处理iot消息异常",ee);
                            }
                        }
                    });
                } catch (Exception e) {
                    logger.info("iot监听出错，第" + cnt + "次");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e1) {

                    }
                    logger.error("", e);
                    e.printStackTrace();
                }
            }
            /*//直链模式
            // 连接配置
            String endPoint = "https://" + aliyunProperties.getAliyunUid() + ".iot-as-http2." + aliyunProperties.getIotRegionId() + ".aliyuncs.com";
            Profile profile = Profile.getAccessKeyProfile(endPoint, aliyunProperties.getIotRegionId(),
                    aliyunProperties.getAliyunAccessKeyId(), aliyunProperties.getAliyunAccessKeySecret());

            MessageClient client = MessageClientFactory.messageClient(profile);
            client.setMessageListener("/" + aliyunProperties.getIotProductKey() + "/#",  messageToken -> {
                Message m = messageToken.getMessage();
                log("receive message from " + m.toString());
                log("topic:" + m.getTopic());
                try {
                    handleMessage(m);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                log("receive : " + new String(messageToken.getMessage().getPayload()));
                return MessageCallback.Action.CommitSuccess;
            });
            client.connect(messageToken -> {
                Message m = messageToken.getMessage();
                log("receive message from connect --- --- " + m.toString());
                log("topic:" + m.getTopic());
                try {
                    handleMessage(m);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                log("receive : " + new String(messageToken.getMessage().getPayload()));
                return MessageCallback.Action.CommitSuccess;
            });*/
        });
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

这里使用了@PostConstruct来实现与程序一并启动，并死循环监听，并使用多线程编程来实现对消息的处理（线程池threadPoolTaskExecutor根据需求配置对应参数）
发布：通过后台处理后的消息根据协议定义转化为的数据封装到消息中发布到队列里，这里最终调用了阿里云com.aliyuncs包中的getAcsResponse(request);来实现发布。
代码（只是发布方法中的其中一个，项目中根据需求有许多不同的发布方法）

public void sendTopic(LsDevice device, byte[] data, IotDeviceShadow.ServerProtocolType protocolType) throws ClientException, AlertException, InterruptedException {
        //编码

        StringBuilder stringBuilder1 = new StringBuilder();
        stringBuilder1.append(String.format("\nIotMessageQueueListener 向设备[" + device.getShowName() + "]发送->的PAYLOAD[%d]:[", data.length));
        for (int i = 0; i < data.length; i++) {
            stringBuilder1.append(String.format("%2X", data[i]));
        }
        stringBuilder1.append("]\n");
        logger.info(stringBuilder1.toString());

        //修安排encoder
        data = java.util.Base64.getEncoder().encode(data);
        sendJob(device, data, protocolType);

        LsDeviceDoor deviceDoor = lsDeviceDoorService.selectOne(
                new EntityWrapper<LsDeviceDoor>()
                        .eq(LsDeviceDoor.DEVICE_ID, device.getId())
                        .orderBy(LsDeviceDoor.ADD_TIME, false)
        );
        DeviceLog deviceLog = new DeviceLog(device, deviceDoor);
        deviceLog.setDesc(protocolType.getType());
        try {
            logHubService.putLog(deviceLog);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
private void sendJob(LsDevice device, byte[] data, IotDeviceShadow.ServerProtocolType protocolType) throws ClientException {
        IotTopic topic = new IotTopic();
        topic.setProductKey(device.getProductKey());

        if (!ValueHelper.isNone(device.getDeviceName())) {
            topic.setDeviceName(device.getDeviceName());
            topic.setTopic(IotDeviceShadow.IotTopicCode.s2d_v1.getTopic());

            PubRequest request = new PubRequest();
            request.setProductKey(aliyunProperties.getIotProductKey());
            request.setMessageContent(Base64.encodeBase64String(data));
            request.setTopicFullName(topic.getTopicString());

            //定义QoS
            if (protocolType.equals(IotDeviceShadow.ServerProtocolType.请求设备开锁)) {
                request.setQos(1); //目前支持QoS0和QoS1
            } else {
                request.setQos(0); //目前支持QoS0和QoS1
            }
            PubResponse response = client.getAcsResponse(request);
            logger.debug(JSONObject.toJSONString(response));
        }
        if (!ValueHelper.isNone(device.getWifiDeviceName())) {
            topic.setDeviceName(device.getWifiDeviceName());
            topic.setTopic(IotDeviceShadow.IotTopicCode.s2w_v1.getTopic());

            PubRequest request = new PubRequest();
            request.setProductKey(aliyunProperties.getIotProductKey());
            request.setMessageContent(Base64.encodeBase64String(data));
            request.setTopicFullName(topic.getTopicString());

            //定义QoS
            if (protocolType.equals(IotDeviceShadow.ServerProtocolType.请求设备开锁)) {
                request.setQos(1); //目前支持QoS0和QoS1
            } else {
                request.setQos(0); //目前支持QoS0和QoS1
            }
            PubResponse response = client.getAcsResponse(request);
            logger.debug(JSONObject.toJSONString(response));
        }
        if (!ValueHelper.isNone(device.getSlaveDeviceName())) {
            topic.setDeviceName(device.getWifiDeviceName());
            topic.setTopic(IotDeviceShadow.IotTopicCode.s2e_v1.getTopic());

            PubRequest request = new PubRequest();
            request.setProductKey(aliyunProperties.getIotProductKey());
            request.setMessageContent(Base64.encodeBase64String(data));
            request.setTopicFullName(topic.getTopicString());

            //定义QoS
            if (protocolType.equals(IotDeviceShadow.ServerProtocolType.请求设备开锁)) {
                request.setQos(1); //目前支持QoS0和QoS1
            } else {
                request.setQos(0); //目前支持QoS0和QoS1
            }
            PubResponse response = client.getAcsResponse(request);
            logger.debug(JSONObject.toJSONString(response));
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

这里需要注意的，我们建立IotDeviceShadow类来作为协议封装，根据其中的枚举类型来对应协议中的消息号，通过枚举类型代替难记的16进制消息号，减少开发难度，提高代码可阅读性。

流水号

在通信过程中，可能会出现不稳定的情况，导致消息的丢失，或者用户的重复操作等，如果没有某个机制来代表整个通信流程的开始以及结束，可能会导致各种各样的问题，导致整个系统乱套。
举个例子：

比如用户要开启某个门柜，通过扫码后台会在数据库中验证设备状态，验证通过后发送请求开门消息给设备，设备会收到消息后开门…
大致过程：扫码-》后台发送开门消息-》设备接收消息-》执行开门
这时候如果由于某种原因，网络不稳定，导致延时消息没有发到设备上，这时候用户有执行扫码，发送了第二条消息，执行了开门操作，用户选购完关门结账离开。而这时候第一条消息如果有发到设备上，导致设备执行开门操作，这将是一场灾难…

解决方案：为了防止这种情况，可以设置一个流水号（process_version）在数据库中，设备也会有该流水号，当后台要发送开门请求时会给这个流水号+1，将流水号一并发送到设备上，这时候设备会验证该流水号是否与自身的匹配，如果匹配则执行开锁操作并将自身的流水号+1，如果不匹配就不执行，这样如果有第二条重复的请求消息过来时，设备就能够识别出来。
总结：这个机制是非常重要的，是保证每个操作流程的有始有终，是实现后台与设备通信的基础。

文章总结：

• 了解到MQTT协议
• 应用到了订阅者模式
• 实现了监听效果，使用了多线程来实现对队列消息中的处理
• 熟悉了订阅和发布的代码过程
• 通过流水号来防止通信问题