资讯详情

MQTT协议

MQTT协议

  • MQTT简介
  • 特性
  • MQTT协议格式
  • 实现方式
  • MQTT的搭建（ubuntu）
  • MQTT权限配置
    • 修改配置文件
    • 添加用户信息
    • 添加Topic与用户的关系
    • 用户认证测试
  • MQTT实现（Java语言）
    • Pom
    • ClientMQTT
    • ServerMQTT
    • PushCallback

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）：
    • QoS0:最多一次TCP/IP网络。消息会丢失或重复。这个级别可以用于以下情况。环境传感器数据丢失一次读取记录并不重要，因为很快就会有第二次发送。这种方法主要是普通的APP如果您的智能设备在新闻推送过程中没有联网，则推送过去没有收到，则无法再次联网。
    • QoS1:至少一次，以确保信息到达，但可能会重复。
    • QoS2:只有一次，以确保信息达到一次。这个级别可以用于一些严格的计费系统。在计费系统中，重复或丢失消息会导致错误的结果。这种最高质量的信息发布服务也可以用于即时通讯APP推送，确保用户只收到一次。
  5. 小传输，费用很小(固定长度的头是2字节)，协议交换最小化，以减少网络流量。

这就是为什么在介绍中说它非常适合在物联网领域，传感器和服务器的通信和信息收集。我们应该知道，嵌入式设备的计算能力和带宽相对较弱，这是非常适合使用该协议传输信息的。

MQTT协议格式

MQTT协议采用二进制数据包，包括固定头、可变头、消息体三个部分；

  • 固定头：存在于所有MQTT在数据包中，长度为2-5字节，包括3部分，数据包类型(4bit），标识位（4bit），数据包剩余长度(1)~4Byte），参考具体含义MQTT协议

  • 可变头：部分MQTT数据包包含

  • 消息体：部分MQTT数据包包含

MQTT在协议中，固定头中的数据包的剩余长度包括可变头和消息体长度，剩余长度可使用1~4Byte表示，4Byte最大可表示256MB（0xFFFFFF7F）。有些数据包没有可变头和消息体，比如PINGREQ数据包只有2Byte，可以计算MQTT数据包的最大值和最小值

  • 最大值：256MB 5Byte，其中256MB剩余长度最大值，包括可变头和消息体；5Byte是固定头，数据包类型(4bit），标识位（4bit），数据包剩余长度(4)Byte）；

  • 最小值：2Byte，比如PINGREQ数据包，只有固定头，没有可变头，消息体；

实现方式

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

MQTT Client: 只需基于设备MQTT协议连接了MQTT Broker ，认为这个设备是MQTT Client, MQTT Client 它可以单独作为出版商和订阅者，也可以作为出版商和订阅者。

MQTT Broker: MQTT Broker 是MQTT协议的核心是接收出版商的信息，然后转发给相应的订阅者。Broker可以对Clinet授权访问并正确Client控制权限。常用C语言编写MQTT Broker 开源库有Mosquitto。如果学习MQTT需要Broker环境可以通过这些开源库自建，也可以由各大云平台提供Broker阿里云、腾讯云等服务。

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

  • Topic，可以理解为新闻类型，订阅者订阅（Subscribe）之后，您将收到主题的消息内容（payload）；
  • payload，可以理解为新闻的内容是指订阅者想要使用的内容。

MQTT的搭建（ubuntu）

1.apt-get安装mqtt相关包

apt-get install mosquitto

在这里插入图片描述 安装客户端

apt-get install mosquitto-clients

2.测试mosquitto是否正确运行

service mosquitto status

3.本机终端测试mqtt

打开一个终端，订阅主题

mosquitto_sub -h 192.168.158.128 -t "mqtt" -v
  • -h 指定要连接的MQTT服务器
  • -t 这是订阅主题mqtt
  • -v 打印更多调试信息

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

mosquitto_pub -h 192.168.158.128 -t "mqtt" -m "hello world"
  • -h 指定要连接的MQTT服务器
  • -t 将信息推送到指定的主题
  • -m 指定消息内容

MQTT权限配置

前面我们基于Mosquitto服务器已设成功，但默认允许匿名用户登录，正式推出的项目需要用户认证（当然，用户通常与数据库映射，但我们将直接将用户写入配置文件）

配置文件参数说明：

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:访问控制列表

修改配置文件

 vi /etc/mosquitto/mosquitto.conf


allow_anonymous false
password_file /etc/mosquitto/pwfile
acl_file /etc/mosquitto/acl

添加用户信息

mosquitto_passwd -c /etc/mosquitto/pwfile zysheep

添加Topic和用户的关系

user zysheep
topic write  mtopic/#

user zysheep
topic read mtopic/#

主题层级分隔符 / : 用于分割主题层级，/分割后的主题，这是消息主题层级设计中很重要的符号。 比方说： aaa/bbb和 aaa/bbb/ccc 和aaa/bbb/ccc/ddd ，这样的消息主题格式，是一个层层递进的关系，可通过多层通配符同时匹配两者，或者单层通配符只匹配一个。 这在现实场景中，可以应用到：公司的部门层级推送、国家城市层级推送等包含层级关系的场景。

单层通配符 +: 单层通配符只能匹配一层主题。比如： aaa/+ 可以匹配 aaa/bbb ，但是不能匹配aaa/bbb/ccc。单独的+号可以匹配单层的所有推送

多层通配符 #: 多层通配符可以匹配于多层主题。比如: aaa/# ,不但可以匹配aaa/bbb，还可以匹配aaa/bbb/ccc/ddd。也就是说，多层通配符可以匹配符合通配符之前主题层级的所有子集主题。单独的#匹配所有的消息主题.

用户认证测试

1.重启Mosquitto步骤,查看mosquitto的进程

ps -aux|grep mosquitto

2. 杀死进程

sudo kill -9 25372

3.重新启动mosquitto

mosquitto -c /etc/mosquitto/mosquitto.conf

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

mosquitto_sub -h 192.168.158.128 -t "mtopic" -v

拒绝连接：未授权

5.订阅端启动（加用户）

mosquitto_sub -h 192.168.158.128 -t "mtopic" -v -u zysheep -P 123456
  • -h 指定要连接的MQTT服务器
  • -t 订阅主题，此处为mqtt
  • -v 打印更多的调试信息
  • -u 设置的用户
  • -P 用户密码

6.发布端启动

mosquitto_pub -h 192.168.158.128 -t mtopic -u zysheep  -p 123456 -m "hello world"

MQTT实现（Java语言）

Pom

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

ClientMQTT

package cn.zysheep.mybatisspringboot.mqtt;

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;
/** * Subscribe 客户端订阅主题 */
public class ClientMQTT { 
        
    public static final String HOST = "tcp://192.168.158.128:1883";
    public static final String TOPIC = "mtopic/zysheep/123";
    private static final String clientid = "client11";
    private MqttClient client;
    private MqttConnectOptions options;
    private String userName = "zysheep";
    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};
            // qos 对消息处理的几种机制。
            // 0 表示的是订阅者没收到消息不会再次发送，消息会丢失。 最多一次，即：<=1
            // 1 表示的是会尝试重试，一直到接收到消息，但这种情况可能导致订阅者收到多次重复消息。 至少一次，即：>=1
            // 2 多了一次去重的动作，确保订阅者收到的消息有一次。 一次，即：=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();
    }
}

ServerMQTT

package cn.zysheep.mybatisspringboot.mqtt;
import cn.zysheep.mybatisspringboot.entity.Wares;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
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;

/** * 服务器向多个客户端推送主题，即不同客户端可向服务端订阅相同的主题 * * Publish 服务端向主题发布消息 */

public class ServerMQTT { 
        
    //tcp://MQTT安装的服务器地址:MQTT定义的端口号
    public static final String HOST = "tcp://192.168.158.128:1883";
    //定义一个主题
    public static final String TOPIC = "mtopic/zysheep/123";
    //定义MQTT的ID，可以在MQTT服务配置中指定
    private static final String clientid = "server11";

    private MqttClient client;
    private MqttTopic topic11;
    private String userName = "zysheep";
    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, JsonProcessingException, InterruptedException { 
        
        ServerMQTT server = new ServerMQTT();

        server.message = new MqttMessage();
        server.message.setQos(1);
        server.message.setRetained(true);
        Wares wares = new Wares();
        wares.setName("常德牌水表");
        wares.setPicAddress("常德");
        ObjectMapper objectMapper = new ObjectMapper();
        String s = objectMapper.writeValueAsString(wares);
        server.message.setPayload(s.getBytes());
        for (int i = 0; i < 10; i++) { 
        
            if (i%2 == 0){ 
        
                Thread.sleep(5000);
            }
            server.publish(server.topic11 , server.message);
        }


        System.out.println(server.message.isRetained() + "------ratained状态");
    }
}

PushCallback

package cn.zysheep.mybatisspringboot.mqtt;

import cn.zysheep.mybatisspringboot.entity.Wares;
import com.fasterxml.jackson.databind.ObjectMapper;
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());
        String s = new String(message.getPayload());
        ObjectMapper objectMapper = new ObjectMapper();
        Wares wares = objectMapper.readValue(s, Wares.class);
        System.out.println(wares);


        //System.out.println("接收消息内容 : " + new String(message.getPayload()));
    }
}

标签： 流量传感器p11231sn

锐单商城拥有海量元器件数据手册、 IC替代型号，打造 电子元器件IC百科大全！

锐单商城 - 一站式电子元器件采购平台