mqtt开发

2021-05-17

一、MQTT简介

　　MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的轻量级协议，该协议构建于TCP/IP协议之上，MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。

　　MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

二、特性

　　MQTT协议工作在低带宽、不可靠的网络的远程传感器和控制设备通讯而设计的协议，它具有以下主要的几项特性：

　　（1）使用发布/订阅消息模式，提供一对多的消息发布，解除应用程序耦合。

　　（2）对负载内容屏蔽的消息传输。

　　（3）使用TCP/IP提供网络连接。

　　主流的MQTT是基于TCP连接进行数据推送的，但是同样有基于UDP的版本，叫做MQTT-SN。这两种版本由于基于不同的连接方式，优缺点自然也就各有不同了。

　　（4）有三种消息发布服务质量：

　　“至多一次”，QoS 0，消息发布完全依赖底层TCP/IP网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。这一种方式主要普通APP的推送，倘若你的智能设备在消息推送时未联网，推送过去没收到，再次联网也就收不到了。

　　“至少一次”，QoS 1，确保消息到达，但消息重复可能会发生。

　　“只有一次”，QoS 2，确保消息到达一次。在一些要求比较严格的计费系统中，可以使用此级别。在计费系统中，消息重复或丢失会导致不正确的结果。这种最高质量的消息发布服务还可以用于即时通讯类的APP的推送，确保用户收到且只会收到一次。　

　　（5）小型传输，开销很小（固定长度的头部是2字节），协议交换最小化，以降低网络流量。

　　这就是为什么在介绍里说它非常适合“在物联网领域，传感器与服务器的通信，信息的收集”，要知道嵌入式设备的运算能力和带宽都相对薄弱，使用这种协议来传递消息再适合不过了。

三、实现方式

　　实现MQTT协议需要客户端和服务器端通讯完成，在通讯过程中，MQTT协议中有三种身份：发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。

　　MQTT传输的消息分为：主题（Topic）和负载（payload）两部分：

　　（1）Topic，可以理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容（payload）；

　　（2）payload，可以理解为消息的内容，是指订阅者具体要使用的内容。

四、MQTT的搭建（ubuntu）

　　1、apt-get安装mqtt相关包

　　2、测试mosquitto是否正确运行

　　3、本机终端测试mqtt

　　打开一个终端，订阅主题

1	mosquitto_sub -h 192.168.1.102 -t "mqtt" -v

　　【-h】指定要连接的MQTT服务器
　　【-t】订阅主题，此处为mqtt
　　【-v】打印更多的调试信息

　　再打开一个终端，发布主题

1	mosquitto_pub -h 192.168.1.102 -t "mqtt" -m "Hello Stonegeek"

　　【-h】指定要连接的MQTT服务器
　　【-t】向指定主题推送消息
　　【-m】指定消息内容

　　结果展示

五、MQTT权限配置

　　前面我们基于Mosquitto服务器已经搭建成功了，但是默认是允许匿名用户登录，对于正式上线的项目则是需要进行用户认证（当然，用户一般都会与数据库映射，不过在这里我们就会直接将用户写入配置文件中）

　　1、Mosquitto服务器的配置文件为/etc/mosquitto/mosquitto.conf，关于用户认证的方式和读取的配置都在这个文件中进行

　　配置文件参数说明：

ID	allow_anonymous	password_file	acl_file	result
1	True(默认)			允许匿名方式登录
2	False	password_file		开启用户验证机制
3	False	password_file	acl_file	开启用户验证机制，但访问控制不起作用
4	True	password_file	acl_file	用户名及密码不为空，将自动进行用户验证且受到访问控制的限制；用户名及密码为空，将不进行用户验证且受到访问控制的限制
5	False			无法启动服务

　　allow_anonymous允许匿名

　　password-file密码文件

　　acl_file访问控制列表

　　2、修改配置文件

　　命令：sudo vi /etc/mosquitto/mosquitto.conf

　　3、添加用户信息

　　命令解释： -c 创建一个用户、/etc/mosquitto/pwfile.example 是将用户创建到 pwfile.example 文件中、admin 是用户名。　

　　　　　　　　同样连续会提示连续输入两次密码。注意第二次创建用户时不用加 -c 如果加 -c 会把第一次创建的用户覆盖。

　　　　　　　　至此两个用户创建成功，此时如果查看 pwfile.example 文件会发现其中多了两个用户。

　　4、添加Topic和用户的关系

　　5、用户认证测试

　　（1）重启Mosquitto步骤　　　　

　　查看mosquitto的进程

　　命令：ps -aux|grep mosquitto

　　（2）杀死进程

　　命令：sudo kill -9 pid

　　（3）启动

　　命令：mosquitto -c /etc/mosquitto/mosquitto.conf

　　（4）订阅端启动（不加用户）

　　订阅端启动（加用户）

　　（5）发布端启动

六、MQTT实现（Java语言）

　　注意：由于我们在上面配置了MQTT的用户权限控制，所以下面的用户只能使用stonegeek登录，否则项目会运行报错，而且我们在上面设置的访问控制列表中只有mtopic主题，所以我们必须使用此主题，否则，订阅者会收不到已发布的主题内容（已经测试过了）

　　下面是我们Java语言实现的MQTT服务的发布/订阅

1、添加Maven依赖

<dependency>
  <groupId>org.eclipse.paho</groupId>
  <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  <version>1.1.1</version>
</dependency>

2、ServerMQTT.class

package com.stonegeek;
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;
/**
 * 服务器向多个客户端推送主题，即不同客户端可向服务端订阅相同的主题
 */
public class ServerMQTT {
    //tcp://MQTT安装的服务器地址:MQTT定义的端口号
    public static final String HOST = "tcp://192.168.1.102:1883";
    //定义一个主题
    public static final String TOPIC = "mtopic";
    //定义MQTT的ID，可以在MQTT服务配置中指定
    private static final String clientid = "server11";

    private MqttClient client;
    private MqttTopic topic11;
    private String userName = "stonegeek";
    private String passWord = "123456";

    private MqttMessage message;

    /**
     * 构造函数
     * @throws MqttException
     */
    public ServerMQTT() 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);

            topic11 = client.getTopic(TOPIC);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     *
     * @param topic
     * @param message
     * @throws MqttPersistenceException
     * @throws MqttException
     */
    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());
    }

    /**
     *  启动入口
     * @param args
     * @throws MqttException
     */
    public static void main(String[] args) throws MqttException {
        ServerMQTT server = new ServerMQTT();

        server.message = new MqttMessage();
        server.message.setQos(1);
        server.message.setRetained(true);
        server.message.setPayload("hello,topic11".getBytes());
        server.publish(server.topic11 , server.message);
        System.out.println(server.message.isRetained() + "------ratained状态");
    }
}

3、ClientMQTT.class

package com.stonegeek;
import java.util.concurrent.ScheduledExecutorService;
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.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class ClientMQTT {
    public static final String HOST = "tcp://192.168.1.102:1883";
    public static final String TOPIC = "mtopic";
    private static final String clientid = "client11";
    private MqttClient client;
    private MqttConnectOptions options;
    private String userName = "stonegeek";
    private String passWord = "123456";

    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 {
        ClientMQTT client = new ClientMQTT();
        client.start();
    }
}

4、PushCallback.class

package com.stonegeek;

import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttMessage;

/**
 * 发布消息的回调类 
 *
 * 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。 
 * 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。 
 * 在回调中，将它用来标识已经启动了该回调的哪个实例。 
 * 必须在回调类中实现三个方法： 
 *
 *  public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。 
 *
 *  public void connectionLost(Throwable cause)在断开连接时调用。 
 *
 *  public void deliveryComplete(MqttDeliveryToken token)) 
 *  接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。 
 *  由 MqttClient.connect 激活此回调。 
 */
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()));
    }
}

5、结果展示

发布消息：

MqttClient client = new MqttClient(“tcp://localhost:1883”, MqttClient.generateClientId());
client.connect();
MqttMessage message = new MqttMessage();
message.setPayload(“Hello world from Java”.getBytes());
client.publish(“iot_data”, message);
client.disconnect();

消费消息：

MqttClient client=new MqttClient(“tcp://localhost:1883”, MqttClient.generateClientId());
client.setCallback( new SimpleMqttCallBack() );
client.connect();

出现的问题：

java.io.EOFException：

这个错误是因为服务器上的和本地的clientID是一样的，当本地的服务跑起来时，需要保证clientID唯一，线上的就会断开连接。

//如果一定要收到离线消息,必须使用确定的id。如果确定就一个clientId,链接的tcp一个,确保自己demo的配置clientId与服务器的clientId不要同一个,会报错丢失连接,而且根本不会意识到是自己跑起来的demo与服务器冲突才导致服务器和自己不断报丢失的

有用时间戳作为标识的，也有用uuid作为标识。

七、mqtt主题

利用MQTT一次订阅多个主题
在做智能家居相关的应用时候，需要利用订阅所有设备的主题。这个时候我们需要利用mqtt的通配符的功能，一个订阅可能包含特殊字符，允许你一次定义多个主题。主题层次分隔符被用来在主题中引入层次。单层通配符和多层通配符只能用于订阅(subscribe)消息而不能用于发布(publish)消息，主题层级分隔符两种情况下均可使用

1.主题层级分隔符/
/ 被用来分割主题树的每一层，并给主题空间提供分等级的结构。当两个通配符在一个主题中出现的时候，主题层次分隔符的使用是很重要的。

2.多层通配符#
#是一个匹配主题中任意层次数的通配符。比如说，如果你订阅了finance/stock/ibm/#，你就可以接收到以下这些主题的消息。
finance/stock/ibm
finance/stock/ibm/closingprice
finance/stock/ibm/currentprice/sdahkj

多层通配符有可以表示大于等于0的层次。因此，finance/#也可以匹配到单独的finance，在这种情况下#代表0层。在这种语境下主题层次分隔符/就没有意义了。因为没有可以分的层次。

#和finance/#都是有效的，但是finance#不是有效的。多层通配符#一定要是主题树的最后一个字符。比如说，finance/#是有效的，但是finance/#/closingprice是无效的。

3.单层通配符+
+只匹配主题的一层。比如说，finance/stock/+匹配finance/stock/ibm和finance/stock/xyz，但是不匹配finance/stock/ibm/closingprice。另外，因为单层通配符只匹配1层，finance/+不匹配finance。
单层通配符可以被用于主题树的任意层级，连带多层通配符。它必须被用在主题层级分隔符/的右边，除非它是指定自己。因此，+和finance/+都是有效的，但是finance+无效。单层通配符可以用在主题树的末端，也可以用在中间。比如说，finance/+和finance/+/ibm都是有效的。

4.主题语法和用法
当你建立一个应用，设计主题树的时候应该考虑以下的主题名字的语法和语义：
主题至少有一个字符长。
主题名字是大小写敏感的。比如说，ACCOUNTS和Accounts是两个不同的主题。
主题名字可以包含空格。比如，Accounts payable是一个有效的主题。
以/开头会产生一个不同的主题。比如说，/finnace与finance不同。/finance匹配”+/+”和/+,但不匹配+
不要在任何主题中包含null（Unicode \x0000）字符。
以下的原则应用于主题树的建造和内容

在主题树中，长度被限制于64k内但是在这以内没有限制层级的数目。
可以有任意数目的根节点；也就是说，可以有任意数目的主题树。

八、服务质量等级

服务质量(QualityofService,QoS)等级是消息发送方与消息接收方之间的协议，对应着消息传递时不同的可靠程度。

MQTT有三种QoS等级：

至多一次（QoS 0）
至少一次（QoS 1）
只有一次（QoS 2）
MQTT中消息的发布和订阅都有QoS等级的设置，我们需要将其分开理解

1.publish的QoS等级

1.1 QoS 0

当client1 publish一条消息时，如果设置的QoS等于0，那么client1只会发送一次这条消息，就算broker没有收到，client1也不会管。对于broker来说，它收到client1 publish的消息，就会PUBLISH给众多客户端，收不到自然也不会PUBLISH。产生的效果：broker最终会PUBLISH 0条或1条消息，即至多一次。

publish

1.2 QoS 1

上面那个例子，如果设置QoS等于1，那么client1发送一条消息，broker收到后会发给client1一个应答PUBACK，client1如果收到应答就会停止发送，否则，继续发送。broker做的事情也很简单，收到一次消息，就将消息PUBLISH给众多客户端，同时发现QoS等级为1，就向刚刚给自己publish消息的client1发送一个响应PUBACK。但是在这种交互中存在两个问题，1：client1 publish的消息有可能broker没能收到，2：broker回复的PUBACK client1没能收到。这样会造成：broker收到消息，将消息PUBLISH出去后，client1没能收到PUBACK，就会继续发送消息，broker就会继续PUBLISH，直到client1收到PUBACK为止。产生效果：broker最终会PUBLISH 至少1条消息，即至少一次。

publish QoS 1

1.3 QoS 2

还是上面的例子，如果设置QoS等于2，那么client1发送一条消息后，broker会回一个PUBREC(publish recevied)（但是这里注意broker虽然已经收到了消息但是它不会立马PUBLISH给众多客户端，还要等待后续步骤）,client1收到PUBREC会回应一个PUBREL(publish release)(即既然你broker收到了，我client1就把这条消息释放了，保存起来太占地方),client1也不会再重复发这条消息了,因为都已经释放这个包了,broker收到PUBREL后会回复client1一个PUBCOMP(publish complete)（即这次消息传递完毕）,broker发完PUBCOMP后，才会将刚才的消息PUBLISH给众多客户端。在PUBCOMP之前的交互中，如若有丢包同样会重传。产生效果：broker最终会PUBLISH 1条消息，不会多于1条，也不会少于1条。

qos：2的情况，客户端A如果pub一条消息，服务器四次交互后保证服务器收到该消息，服务器再转发给sub的客户端B，也会有四次交互保证该消息一定转发成功。不成功的话会重试直到成功，多条消息就会多次重试。

publish QoS 2

2021-12-13T08:32:18: Received PUBLISH from GCSM21B0002C (d0, q2, r0, m4, 'g_Mini_GateWay/GCSM21B0002C', ... (84 bytes))
2021-12-13T08:32:18: Sending PUBREC to GCSM21B0002C (m4, rc0)
2021-12-13T08:32:18: Received PUBREL from GCSM21B0002C (Mid: 4)
2021-12-13T08:32:18: Sending PUBCOMP to GCSM21B0002C (m4)

2021-12-13T08:32:18: Sending PUBLISH to ddba815923d946869d4e03e4b3507186 (d0, q2, r0, m17772, 'g_Mini_GateWay/GCSM21B0002C', ... (84 bytes))
2021-12-13T08:32:18: Received PUBREC from ddba815923d946869d4e03e4b3507186 (Mid: 17772)
2021-12-13T08:32:18: Sending PUBREL to ddba815923d946869d4e03e4b3507186 (m17772)
2021-12-13T08:32:18: Received PUBCOMP from ddba815923d946869d4e03e4b3507186 (Mid: 17772, RC:0)

应用场景：

为什么要有这三种服务质量等级呢？我们结合具体需求来思考一下
QoS 0 适合什么场景呢？适合那种消息很频繁但又不那么重要的应用，如每隔几分钟上报一次气温，数据丢了就丢了，反正待会还会上报，并且丢失一次数据也不会产生多大影响。
但是，如果消息对你很重要，你宁愿收到重复的也不愿意丢失一条，那你就采用QoS 1
但是，如果像支付类的应用也采用QoS 1的话，会产生重复支付的问题（其实已经支付成功了，只是手机没有收到回应，又再次发起支付），这时候就需要使用QoS 2了，它保证有且仅有一次消息成功传递。

2.subscribe的QoS等级

说完了publish的QoS，我们再来谈谈subscribe的QoS。其实我们完全可以使用上面三个例子的分析，因为clientA、clientB、clientC 订阅 broker的消息就相当于broker发布消息给他们（这点从后面的抓包分析也可得以验证），我们仍然可用publish的那一套去理解subscribe的QoS，即client A subscribe消息时设置的QoS等于0，那么broker最多只会发一次消息给clientA。QoS设置为1，那么broker至少会发一次消息给到clientA，给不到就重发。QoS设置为2，那么broker保证有且只有一次消息给到clientA。
但是有一个问题需要注意一下，对于同一个topic，如果client1以QoS 2等级发布，clientA以QoS 0等级订阅，那么clientA最多只会收到一条消息，即两者质量等级取最低。同样的，如果clinet1以QoS 0发布一条消息，clientA以QoS 2等级去订阅，clientA仍然最多只会收到一条消息，因为原则是两者QoS取最低。所以如果要想达到真正的QoS 2质量，就要保证publish和subscribe都设置为 QoS 2

九、保留消息 retained

如果PUBLISH消息的RETAIN标记位被设置为1，则称该消息为“保留消息”

Broker会存储每个Topic的最后一条保留消息及其Qos，当订阅该Topic的客户端上线后，Broker需要将该消息投递给它

可以让新订阅的客户端得到发布方的最新的状态值，而不必要等待发送

删除：

方式1：发送空消息体的保留消息；

方式2：发送最新的保留消息覆盖之前的（推荐）