最近公司要做推送平台，我接手做这个任务。在使用过程中遇到一些坑，这里分享给大家。

一.MQTT协议介绍

推送平台一般是基于轻量级的mqtt协议搭建的。mqtt协议是物联网领域常用的，是实现长链接的一种方式。

先介绍一下http协议，使用http协议发起一次请求的过程是：客户端向服务端发起请求并建立链接，服务端处理客户端的请求并响应给客户端，最后客户端和服务端断开连接；也就是说http协议是请求响应式的，响应完成后则断开连接，是一种短链接。

和http协议相比，mqtt协议是一种长链接，一般是发布（pub）/订阅（sub）模式，需要一个broker服务器作为服务端，多个客户端都可以连接在这个broker服务器上，而且客户端和服务端一直保持连接（心跳）不断开，所以是一种长链接。A客户端向broker的某个topic上发送（pub）一条信息，如果有其他的客户端B1-B3订阅（sub）了这个broker上的这个topic的信息，这些客户端就可以收到A客户端发送的信息。

对于broker服务端来说，由于不同的设备都要和mqtt服务器保持心跳，所以如果客户端的数量达到一定规模，broker服务器的压力会非常大。同时，对于客户端来说，一直要和服务端保持心跳，对客户端的耗电量也有挑战。mqtt协议是物联网领域常用协议，原因是，物联网设备（如共享单车的智能锁）有很多时候会出现在网络环境比较差的场景中，设备断连后会不断的尝试和broker发起连接请求。mqtt协议满足这种低耗电量和网络差的场景需求。

对于搭建mqtt的broker的推送平台的技术选型：

1.mosquito，轻量级，单机版，不适合大规模企业级应用，很适合个人学习使用

2.EMQ，适合企业级应用，使用Erlang语言编写，个人感觉语言开发语言小众，后期维护优化成本比较高。当然可以花钱买官方的服务。

3.阿里云，企业级服务推荐使用，花钱开通服务就好了。文档：MQTT 快速入门 - 微消息队列MQTT版 - 阿里云购买地址：微消息队列MQTT版_mqtt协议_直播互动_容器与中间件-阿里云

这三种我都使用过，个人测试使用过mosquito和EMQ。企业初期我们生产环境也使用过EMQ,后面就转移到了阿里云上。

下面介绍以下大家最关心的搭建和测试，这里我是在centos7上搭建的EMQ。

二.EMQ搭建

如果没有服务器，可以使用买阿里云ECS服务器，限时九折，很便宜：云小站_专享特惠_云产品推荐-阿里云

搭建比较容易，这里就不介绍了，有兴趣的可以看看官网，我参考这篇博客：EMQ百万级MQTT消息服务(介绍和搭建)丶Java教程网-IT开发者们的技术天堂

EMQ由于是erlang语言写的，所以操作系统上要求安装erlang环境，并且对erlang的版本号有要求，搭建EMQ建议使用centos7以上的版本，不用升级操作系统的erlang库；在centos7以下的操作系统上搭建emq有个大坑，就是需要升级erlang的库，非常麻烦。

三.emq的管理平台

搭建完成后可以登录管理平台，需要注意下面的端口和协议 默认用户: admin，密码：public (可在平台中配置用户)

图中的协议和端口是对应的，千万不要用错了。

用java代码连接时，我一般使用tcp协议，其他协议暂时没有测试过，有兴趣可以自己测试。

写这篇博客时，我们的场景消费礼品卡（B扫C）：用户出示微信小程序中的礼品卡消费二维码，商家扫码消费，消费完成后要及时推送到用户的小程序上，让用户及时知道消费结果。这里就需要用户小程序出示二维码的同时（sub）监听broker中由后台服务器消费成功后pub的消费结果，微信的开发配置中要求使用websocket的ws和wss协议。ws和wss的差异类似于http和https，表示是否是加密传输。

四.测试pub和sub

测试可以使用客户端工具，也可以使用java代码，也可以使用js代码测试。最简单的就是使用上面的管理平台中的websocket菜单中的工具测试。

1.使用客户端工具

客户端工具有很多，这里推荐一个MQTT.fx,下载地址：Download free MQTT.fx 1.0

具体使用就不介绍了，大家自行网上查找。这个工具可以测试tcp连接，也可以测试ws wss连接，注意不同的协议对应不同的端口

2.java代码测试（这里只测试tcp的，使用1883端口）

官网给的参考代码java：https://www.eclipse.org/paho/clients/java/?spm=a2c4g.11186623.2.8.7b61de4eccPlSt

阿里云支持的各种SDK：SDK下载 - 微消息队列MQTT版 - 阿里云

主要有以下三个文件：

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::ERR_CERT_AUTHORITY_INVALID，因为证书是自签的，浏览器不认，解决办法：在浏览器上请求https://192.168.1.200:8084/mqtt((如果是经过ngnix转发的使用https://www.wuxiaolong.com/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.onload= 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', 'wwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwww')console.log("订阅成功")}})})//监听mq的返回client.on('message', function (topic, message, packet) {// message is Bufferconsole.log("packet", packet.payload.toString())client.end()})}   </script>    </head><body></body>
</html>

刷新浏览器：

这里面的一个最大的坑是wss的使用，因为我们是做微信小程序，微信小程序的接入标准是要求使用安全的网络传输，所以必须使用wss,而且要使用域名和443端口，域名要备案过，所以小程序不能直接访问emq集群，中间必须加一层nginx转发：

我的处理是:

a.小程序使用代码const client = mqtt.connect('wss://www.wuxiaolong.com/mqtt', options)

b.nginx将小程序的请求转换成ws的8083端口，nginx配置转发（nginx需要more_clear_headers模块，配置添加参见：openresty/headers-more-nginx-module）

nginx的配置如下：

server {

listen 80; # ws 绑定到 80 端口用于代码const client = mqtt.connect('ws://www.wuxiaolong.com/mqtt', options)
listen 443 ssl; # wss 绑定到 443 端口用于代码const client = mqtt.connect('wss://www.wuxiaolong.com/mqtt', options)

server_name www.wuxiaolong.com;
ssl on;
ssl_certificate ssl/wuxiaolong.com.cer;
ssl_certificate_key ssl/wuxiaolong.com.key;

location /mqtt {
proxy_redirect off;
proxy_pass http://192.168.1.200:8083; # 反向代理到 EMQ 非加密 WebSocket ws

proxy_set_header Host $host;

proxy_set_header X-Real_IP $remote_addr; # 反向代理保留客户端地址
proxy_set_header X-Forwarded-For $remote_addr:$remote_port;

# WebSocket 额外请求头
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection “upgrade”;
}
}

参考博客：

使用 WebSocket 客户端连接 MQTT 服务器 - 简书

EMQ百万级MQTT消息服务(介绍和搭建)丶Java教程网-IT开发者们的技术天堂

EMQ踩坑之路-搭建/测试/nginx配置websocket的ws及wss/web页面测试连通性、微信小程序使用wss--MQTT推送相关推荐

【微信小程序】模板消息推送（测试成功）。
note: 这几天想做微信小程序消息推送的功能,然后在网上找了好多教程,都没解决问题,官方文档写的教程由不够详细,所以走了好多弯路. 现在问题解决了,记录一下. 我在网上找的好多教程都不详细,虽然是贴 ...
怎么测试本地网页在不同分辨率下电脑显示效果_干货：微信小程序测试过程中的各个要点...
文章转自:新梦想软件测试小程序跟app一样,上线需要经过微信的审核.小程序产品的版本类型分为:开发版.体验版.正式版.开发版.体验版无需审核,只需要给微信号权限,经过扫小程序的二维码才能访问. 单纯 ...
[Unity]腾讯SDK踩坑之路（2）--配置Xcode工程（MSDK和米大师配置代码冲突）
前不久接完MSDK,又花了几天时间改之前的打包工具和MSDK的一些代码,主要是这边有骚操作,需要不带SDK的包和带SDK的包,还有很多杂七杂八的包.然后顺带着打了个iOS测试包,检测SDK有没有接好. ...
微信小程序服务端开发遇到的一些坑
微信小程序服务端开发遇到的一些坑开发环境问题集合小程序不支持基于进入客服事件推送消息小程序不支持永久素材 RestTemplate无法解析上传图片素材的返回 tips 其他小问题开发环境基 ...
微信小程序从入坑到放弃二十二：完美兼容安卓和ios手机的底部评论框
摘要: 不管是在普通的H5页面还是在微信小程序中,底部输入框一直是一大难题,因为键盘会在某些情况下遮挡信输入框一部分!值得庆幸的是:在微信小程序中,我们可以通过监听bindfocus来获取键盘的高度, ...
记坑：mqtt+微信小程序。
背景:使用mpvue搭建微信小程序项目,利用mqtt进行物联网. 之前利用mpvue已经能连上阿里云的物联网平台进行通讯,但是我又搭建了EMQX在自己服务器上,想用自己的服务器进行通讯.首先先用页面写 ...
微信小程序开发系列一：微信小程序的申请和开发环境的搭建
我最近也刚刚开始微信小程序的开发,想把我自学的一些心得写出来分享给大家. 这是第一篇,从零开始学习微信小程序开发.主要是小程序的注册和开发环境的搭建. 首先我们要在下列网址申请一个属于自己的微信小程序 ...
微信小程序开发系列一：微信小程序的申请和开发环境的搭建 1
我最近也刚刚开始微信小程序的开发,想把我自学的一些心得写出来分享给大家. 这是第一篇,从零开始学习微信小程序开发.主要是小程序的注册和开发环境的搭建. 首先我们要在下列网址申请一个属于自己的微信小程序 ...
从零开始快速搭建SpringBoot+Mybatis+小程序应用--微信小程序的入门和前后端的联调
目录项目介绍 vx小程序简介 VX开发工具介绍列表页开发 list的编写列表页前后端联调信息编辑页开发 operation的编写区域信息编辑页的联调项目介绍从0搭建后端的Springbo ...

EMQ踩坑之路-搭建/测试/nginx配置websocket的ws及wss/web页面测试连通性、微信小程序使用wss--MQTT推送