MQTT 的搭建、测试、应用及小程序的集成！

关注
发布于: 2020 年 07 月 16 日
前几天在InfoQ上发了一篇B站入职离职的文章，被平台好心的编辑选为热门文章放到首页了，这几天在InfoQ上经常会给我小伙伴给我点赞、问我问题，这个时候我的头像左边的小铃铛就会出现红色的通知消息。这个通知是怎么做的呢？
﻿
如果我已经打开了InfoQ平台的首页，看到有4个通知消息，这个时候又有个小伙伴给我点了个赞，在我没有刷新页面，我能马上知道现在我有5条通知消息吗？
﻿
虽然我不知道InfoQ的内部实现机制是怎样的，但是大概是这几种方式：
﻿
这个通知不会自动刷新。只有在我每次刷新页面的时候，会到数据库查询一下我有几条未读消息，显示在铃铛图标上。在页面未刷新的情况下，即使有小伙伴给我点赞、关注、收藏，也不会更新未读消息的条数。
这个通知会自动刷新，但是会有延时。前端页面在js中写个定时任务，每隔一段时间，调用ajax查询一下数据库，统计未读消息的数量，然后更新铃铛图标上的未读消息的数字。
这个通知会自动实时刷新。这就需要后端服务器主动通知前端页面了。是不是有很多小伙伴没有做过这种后台主动通知前端的业务，没关系，这就是我今天要说的内容了。
﻿
﻿
从InfoQ的这个功能来说，没有实时更新数据用户也是可以接受的。但是如果是支付业务就不一样了，用户支付后，跳到支付结果页，如果支付结果还是支付中，为了以最快的速度让用户知道支付结果提高用户体验，那么就需要前端页面及时轮询查询后端或者后端通知前端页面最终的支付结果了。
﻿
这里就给大家介绍一种物联网里的常用协议MQTT，以及MQTT的使用。
﻿
一.MQTT协议介绍﻿
推送平台一般是基于轻量级的mqtt协议搭建的。mqtt协议是物联网领域常用的，是实现长链接的一种方式。
﻿
大家都知道http协议，使用http协议发起一次请求的过程大致分四步：客户端和服务端建立链接，客户端和服务端发送请求并，服务端处理客户端的请求并响应给客户端，最后客户端和服务端断开连接；也就是说http协议是请求响应式的，响应完成后则断开连接，是一种短链接。
﻿
和http协议相比，mqtt协议是一种长链接，一般是发布（pub）/订阅（sub）模式，需要一个broker服务器作为服务端，多个客户端都可以连接在这个broker服务器上，而且客户端和服务端一直保持连接（心跳）不断开，所以是一种长链接，即使断开了，客户端还会不断尝试继续连上服务端。管理客户端向broker的某个topic上发送（pub）一条信息，如果有其他的客户端订阅（sub）了这个broker上的这个topic的信息，这些客户端就可以收到管理客户端发送的信息。
﻿
比如物联网设备共享单车，开锁的过程就是这样。所有的共享单车都会通过智能锁里面的物联网卡和信号基站相连，信号基站会和机房里的mqtt broker服务器连接。这时智能锁的就会订阅broker上的某个topic。当我们打开手机上共享单车的APP，扫码开锁时，App就会向Broker的这辆车订阅的topic上发送一条开锁指令，这时共享单车就会及时收到开锁指令并开锁。
﻿
如果使用这种技术，一个小伙伴在InfoQ上给我的文章点赞，消息到达broker，这时我使用浏览器提前登录InfoQ，马上就监听到有用户给我发通知，并显示在小铃铛上面。
﻿
﻿
对于broker服务端来说，由于不同的设备都要和mqtt broker保持心跳，所以如果客户端的数量达到一定规模，broker服务器的压力会非常大。同时，对于客户端来说，一直要和broker保持心跳，对客户端的耗电量也有挑战。mqtt协议是物联网领域常用协议，原因是物联网设备（如共享单车）有很多时候会出现在网络环境比较差的场景中，设备断连后会不断的尝试和broker发起连接请求，网络环境差也会更耗电。而mqtt协议满足这种低耗电量和网络差的场景需求。
﻿
对于搭建mqtt的broker的推送平台的技术选型：
﻿
mosquito，轻量级，单机版，不适合大规模企业级应用，很适合个人学习使用。
﻿
EMQ，适合企业级应用，使用Erlang语言编写，个人感觉语言开发语言小众，后期维护优化成本比较高。当然可以花钱买官方的服务。
﻿
阿里云，企业级服务推荐使用，花钱开通服务就好了。文档：https://help.aliyun.com/document_detail/44867.html?spm=a2c4g.11186623.4.2.3d64de4e1acgY2  
﻿
 这三种我都使用过，个人测试使用过mosquito和EMQ。企业初期我们生产环境也使用过EMQ,后面就转移到了阿里云上。
﻿
下面介绍以下大家最关心的搭建和测试，这里我是在centos7上搭建的EMQ。
﻿
二.EMQ搭建﻿
EMQ搭建比较容易，这里就不介绍了，有兴趣的可以看看官网，我参考这篇博客：https://www.liangzl.com/get-article-detail-719.html
﻿
EMQ由于是erlang语言写的，所以操作系统上要求安装erlang环境，并且对erlang的版本号有要求，搭建EMQ建议使用centos7以上的版本，不用升级操作系统的erlang库；在centos7以下的操作系统上搭建emq有个大坑，就是需要升级erlang的库，非常麻烦。
﻿
也可以搭建轻量级的mosquito，这个搭建比较简单。
﻿
三.EMQ的管理平台﻿
搭建完成后可以登录管理平台，需要注意下面的端口和协议  默认用户: admin，密码：public (可在平台中配置用户)
﻿
图中的协议和端口是对应的，千万不要用错了。
﻿
用java代码连接时，我一般使用tcp协议。在做微信小程序或支付宝小程序时，可能要使用wss协议。
﻿
写这篇博客时，我们的场景消费礼品卡（B扫C）：用户出示微信小程序中的礼品卡消费二维码，商家扫码消费，消费完成后要及时推送到用户的小程序上，让用户及时知道消费结果。这里就需要用户小程序出示二维码的同时监听（sub）broker中由后台服务器消费成功后pub的消费结果，微信的开发配置中要求使用websocket的ws和wss协议。ws和wss的差异类似于http和https，表示是否是加密传输。
四.测试pub和sub﻿
测试可以使用客户端工具，也可以使用java代码，也可以使用js代码测试。最简单的就是使用上面的管理平台中的websocket菜单中的工具测试。
﻿
1.使用客户端工具﻿
客户端工具有很多，这里推荐一个MQTT.fx,下载地址：https://download.freedownloadmanager.org/Windows-PC/MQTT-fx/FREE-1.0.html
﻿
具体使用就不介绍了，大家自行网上查找。这个工具可以测试tcp连接，也可以测试ws wss连接，注意不同的协议对应不同的端口
2.java代码测试（这里只测试tcp的，使用1883端口）﻿
官网给的参考代码java：https://www.eclipse.org/paho/clients/java/?spm=a2c4g.11186623.2.8.7b61de4eccPlSt
﻿
阿里云支持的各种SDK：https://help.aliyun.com/document_detail/44866.html?spm=a2c4g.11186623.2.32.6800699bTY6jNl#concept-44866-zh
﻿
主要有以下三个文件：
﻿
﻿
maven坐标：
﻿
       <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.1.0</version>
        </dependency>
﻿
PushCallBack.java文件：
﻿
package mqtt;
 
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
 
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
 
 
public class PushCallBack implements MqttCallback{
 
    public void connectionLost(Throwable cause) {
          // 连接丢失后，一般在这里面进行重连
          System.out.println("连接断开，可以做重连");
    }
 
    public void deliveryComplete(IMqttDeliveryToken token) {
          System.out.println("deliveryComplete---------" + token.isComplete());
    }
 
    public void messageArrived(String topic, MqttMessage message) throws Exception {
          // subscribe后得到的消息会执行到这里面
          System.out.println("接收消息主题 : " + topic);
          System.out.println("接收消息Qos : " + message.getQos());
          System.out.println("接收消息内容 : " + new String(message.getPayload()));
    }
 
}
﻿
Client.java文件：
﻿
package mqtt;
 
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
 
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 
public class Client {
 
    public static final String HOST = "tcp://192.168.1.200:1883";
    public static final String TOPIC = "toclient/124";
    private static final String clientid = "client124";
    private MqttClient client;
    private MqttConnectOptions options;
    private String userName = "admin";
    private String passWord = "public";
 
    private ScheduledExecutorService scheduler;
 
    private void start() {
        try {
            // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client = new MqttClient(HOST, clientid, new MemoryPersistence());
            // MQTT的连接设置
            options = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(true);
            // 设置连接的用户名
            options.setUserName(userName);
            // 设置连接的密码
            options.setPassword(passWord.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(10);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(20);
            // 设置回调
            client.setCallback(new PushCallBack());
            MqttTopic topic = client.getTopic(TOPIC);
            //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
            options.setWill(topic, "close".getBytes(), 2, true);
 
            client.connect(options);
            //订阅消息
            int[] Qos  = {1};
            String[] topic1 = {TOPIC};
            client.subscribe(topic1, Qos);
 
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
 
    public static void main(String[] args) throws MqttException {
        Client client = new Client();
        client.start();
    }
}
﻿
Server.java文件：
﻿
package mqtt;
 
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 
import java.io.*;
 
public class Server {
 
    public static final String HOST = "tcp://192.168.1.200:1883";
    public static final String TOPIC = "toclient/124";
    public static final String TOPIC125 = "toclient/125";
    private static final String clientid = "server";
 
    private MqttClient client;
    private MqttTopic topic;
    private MqttTopic topic125;
    private String userName = "admin";
    private String passWord = "password";
 
    private MqttMessage message;
 
    public Server() throws MqttException {
        // MemoryPersistence设置clientid的保存形式，默认为以内存保存
        client = new MqttClient(HOST, clientid, new MemoryPersistence());
        connect();
    }
 
    private void connect() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(false);
        options.setUserName(userName);
        options.setPassword(passWord.toCharArray());
        // 设置超时时间
        options.setConnectionTimeout(10);
        // 设置会话心跳时间
        options.setKeepAliveInterval(20);
        try {
            client.setCallback(new PushCallBack());
            client.connect(options);
            topic = client.getTopic(TOPIC);
            topic125 = client.getTopic(TOPIC125);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
 
    public void publish(MqttTopic topic , MqttMessage message) throws MqttPersistenceException,
            MqttException {
        MqttDeliveryToken token = topic.publish(message);
        token.waitForCompletion();
        System.out.println("message is published completely! "
                + token.isComplete());
    }
 
    public static void main(String[] args) throws MqttException {
        Server server = new Server();
 
        server.message = new MqttMessage();
        server.message.setQos(2);
        server.message.setRetained(true);
//        server.message.setPayload("给客户端124推送的信息".getBytes());
 
        byte[] sends=File2Bytes("D://test.rar");
 
        server.message.setPayload(sends);
        server.publish(server.topic , server.message);
 
//        server.message = new MqttMessage();
//        server.message.setQos(2);
//        server.message.setRetained(true);
//        server.message.setPayload("给客户端125推送的信息".getBytes());
//        server.publish(server.topic125 , server.message);
 
        System.out.println(server.message.isRetained() + "------ratained状态");
    }
}
﻿
3.js代码测试（使用的是ws和wss协议）﻿
mqtt的js客户端网上有很多，常用的两个是Paho.mqtt.js和MQTT.js，有兴趣的可以去github上详细了解，这里只说MQTT.js。
﻿
我的测试是在chrome浏览器中做的，所以需要html页面和js文件。测试步骤如下
﻿
﻿
a.下载MQTT.js .打开下面的连接，将浏览器中的js代码放到新建的mqtt.min.js文件中。
﻿
https://unpkg.com/mqtt/dist/mqtt.min.js
﻿
b.在html页面中引入这个js。由于是简单测试，这个html页面很简单，是个空白页面，通过浏览器的调试窗口（F12）的console可以看到输出结果。
﻿
这里有两个注意点：使用ws协议必须使用emq的8083端口，必须在请求地址端口后面添加"/mqtt“,  ws://192.168.1.200:8083/mqtt ,  wss://192.168.1.200:8084/mqtt; 第一次链接时，浏览器（这里测试的是chrom浏览器，根据浏览器的厂商或者浏览器的不同版本会有差异）提示failed: Error in connection establishment: net::ERRCERTAUTHORITY_INVALID，因为证书是自签的，浏览器不认，解决办法：在浏览器上请求https://192.168.1.200:8084/mqtt,会提示不安全，选择高级，继续访问，就解决了这个问题。操作完成后再刷新该testMqtt.html页面就链接成功了。
﻿
testMqtt.html的页面如下：
﻿
<!DOCTYPE html>
 <html>
     <head>
         <meta charset="utf-8">
         <title></title>
         <script src="./mqtt.min.js" type="text/javascript"></script>
         <script type="text/javascript">
         
    
             window.οnlοad= function connectMq(){
 
    // 连接选项
    const options = {
      connectTimeout: 4000, // 超时时间
      // 认证信息 按自己需求填写
      clientId: 'clientId123456',
      username: 'admin',
      password: 'public',
    }
 
    //a.使用ws协议必须使用emq的8083端口，必须在请求地址端口后面添加"/mqtt"
    //const client = mqtt.connect('ws://192.168.1.200:8083/mqtt', options)
    //const client = mqtt.connect('ws://www.wuxiaolong.com/mqtt', options)
    
    
    //b.使用wss协议必须使用emq的8084端口，必须在请求地址端口后面添加"/mqtt"；
    //第一次链接时，浏览器（这里测试的是chrom浏览器，根据浏览器的厂商或者浏览器的不同版本会有差异）提示failed: Error in connection establishment: net::ERR_CERT_AUTHORITY_INVALID，
    //因为证书是自签的，浏览器不认，解决办法：在浏览器上请求https://192.168.1.200:8084/mqtt((如果是经过ngnix转发的使用https://www.wuxiaolong.com/mqtt)),会提示不安全，选择高级，继续访问，就解决了这个问题。操作完成后再刷新该testMqtt.html页面就链接成功了。
    //const client = mqtt.connect('wss://192.168.1.200:8084/mqtt', options)
    const client = mqtt.connect('wss://www.wuxiaolong.com/mqtt', options)
    
    
    client.on('reconnect', (error) => {
      console.log('正在重连:', error)
    })
 
    client.on('error', (error) => {
      console.log('连接失败:', error)
    })
 
    client.on('connect', (e) => {
      console.log('成功连接服务器111')
        //订阅一个主题
      client.subscribe('wuxiaolong/test/topic', { qos: 0 }, function (err) {
        if (!err) {
          client.publish('wuxiaolong/test/topic', 'wwwwwwwwwwwwwwwwwwww')
          console.log("订阅成功")
        }
 
      })
    })
    //监听mq的返回
    client.on('message', function (topic, message, packet) {
      // message is Buffer
      console.log("packet", packet.payload.toString())
      client.end()
    })
 
}   
 
         </script>    
     </head>
     <body>
     </body>
</html>
﻿
刷新浏览器：
﻿
这里面的一个最大的坑是wss的使用，因为我们是做微信小程序，微信小程序的接入标准是要求使用安全的网络传输，所以必须使用wss,而且要使用域名和443端口，域名要备案过，所以小程序不能直接访问emq集群，中间必须加一层nginx转发：
﻿
我的处理是: 
1.1.小程序使用代码
﻿
const client = mqtt.connect('wss://www.wuxiaolong.com/mqtt', options)
﻿
1.2.nginx将小程序的请求转换成ws的8083端口，nginx配置转发（nginx需要moreclearheaders模块）
﻿
如果配置Nginx比较麻烦，大家可以使用内网内网穿透工具如Ngrok等，进行内网穿透。
﻿
﻿
【传播知识，共享价值】，感谢小伙伴们的关注和支持，我是【诸葛小猿】，一个彷徨中奋斗的互联网民工。
﻿
﻿