MQTT协议

MQTT协议是一个应用层协议，他要求使用的传输层协议能提供有序的，可靠的双向字节流传输服务。

MQTT协议通信对象分为客户端和服务端，数据的传输以消息为单位，每个消息包含主题，消息服务质量和有效数据。消息的发送和接收都需要依赖主题：A订阅主题T1，B订阅主题T2，A想发送消息给B，就需要将消息发送到主题T2。

一，客户端

通信双方的身份。mqtt通信双方都是客户端，消息传递是以订阅主题的形式。

• 发布消息到服务端以转发到其他客户端
• 订阅主题以接收其他客户端的消息
• 发起连接/断开

二，服务端

作为客户端与客户端之间通信的中介。

• 接收客户端发布的消息
• 接收客户端的连接
• 转发消息给特定的客户端
• 接收客户端的订阅主题请求

三，消息服务质量

消息服务质量决定了一个MQTT报文的消息应如何处理。

• qos0：客户端仅讲报文发送一次，服务端不需要应答，报文是否送达不保证。
• qos1：客户端发送报文，服务端接收到后必须应答，超时重发。
• qos2：客户端发送报文，服务端接收到后应答，客户端再次发送一个release报文，服务端发送该报文到指定主题下的客户端，并应答。

qos0-qos2，服务的质量逐步提高。qos2能使客户端对报文的发送有全局的掌控。qos0很不负责任，嫁出去的女儿泼出去的水，不管了。qos1还行。

四，主题

MQTT主题是一个字符串，由服务端维护。客户端通过在服务端订阅主题，可接收来自该主题的消息。客户端可向该主题发送消息，所有订阅该主题的客户端都会受到该消息。实现的方式是：客户端向服务端发送带主题的消息到服务端，由服务端将消息转发给订阅该主题的客户端，所以服务端地位重要。

MQTT主题支持分级，如/Home/BathRoom/Mirror，/Home/LivingRoom/Tv，是同一等级下的主题。

主题也支持通配符#，发送消息到主题 /Home/# 则BathRoom和LivingRoom主题下的客户端都会收到消息。

主题也支持单层通配符+，/Home/+ 则BathRoom和LivingRoom主题下的客户端会受到消息，但/Home/BathRoom/Mirror和/Home/LivingRoom/Tv 不会收到消息，这点需要跟#区别。

五，MQTT控制报文

MQTT协议依靠MQTT控制报文来通信

5.1 固定报头

表示控制报文的类型，报文的一些标志位（包括消息服务质量），以及报文剩余的字节长度。

5.2 可变报头

可变报头的内容跟报文类型有关。

包含报文标识符，一个16bit的数据，用于唯一的标记此次通信的报文。当客户端处理完当前报文后，标识符可释放重用。qos0的消息不需要标识符，因为不需要服务端的应答。

5.3 有效载荷

前面都是协议规定必须的，有效载荷是真正的用户数据。不同类型的报文，有效载荷里的数据不同。

5.4 控制报文类型

不同类型的报文，其不同点在于可变报头及有效载荷。就以这两部分看看主要的报文：

5.4.1 连接报文

• 可变报头：包含 “使用传输层协议名”（TCP），MQTT协议等级（3.1.1）,连接标志，保持连接。

连接标志：指示有效载荷部分的内容。

保持连接：MQTT客户端需要在一个时间内给服务端发送心跳报文，服务端也要在规定时间内应答。由此判断通信双方是否在线。

• 有效载荷

由连接标志指示其内容，出现的顺序为：**客户端标识符，遗嘱主题，遗嘱消息，**用户名，密码。

客户端标志：字符串，用于唯一标志一个客户端。

遗嘱：是一种错误补救方式，当客户端以异常方式断开连接时，服务端长时间未能联系到客户端，则服务端将该客户端的遗嘱消息发布到遗嘱主题。

用户名和密码：字符串，用于识别客户端是否合法。

5.4.2 发布publish

• 固定报头：
  
  DUP：重发标志，是否是一个重发的报文，qos0大咩。

RETAIN:保留标志位。若为1，服务端需要在内存中保留该消息。

• 可变报头：主题及报文标识符
• 有效载荷：应用数据，一般是json字符串。

5.4.3 订阅主题subscribe

可订阅多个主题

• 有效载荷：包含一个主题过滤器表示客户端要订阅的主题，消息服务质量qos。主题过滤器是一个字符串，后面跟着一个字节的消息服务质量，组成一组，一个订阅报文可包含多组这样的东西，来支持订阅多个主题。

六，安全

安全的实现主要依赖于传输层协议，推进TLS协议服务端使用8883端口。

轻量加密AES

ESP-MQTT API 指南

概括

ESP-MQTT是一个MQTT协议客户端的应用程序

一，特性

• 支持多种传输层协议如：TCP,SSL,Websocket,wws.
• 使用url建立连接
• 允许一个应用中多个客户端
• 支持订阅，发布，认证，遗嘱，保活和3个消息质量

二，应用示例

• protocols/mqtt.tcp:使用tcp，1883 端口
• protocols/mqtt/ssl:使用tcp，端口8883，比较安全
• protocols/mqtt/ssl_psk:使用tcp，基于公钥加密认证，端口8883
• protocols/mqtt/ws:使用websocket，端口80
• protocols/mqtt/wss:使用wss，端口443

三，初始化配置

3.1 URI

• 当前支持mqtt，mqtts，ws，wss方式

• mqtt 使用tcp例子：

  • mqtt://mqtt.eclipse.org: MQTT over TCP, default port 1883
  • mqtt://mqtt.eclipse.org:1884 MQTT over TCP, port 1884
  • mqtt://username:password@mqtt.eclipse.org:1884 MQTT over TCP, port 1884, with username and password

• MQTT over SSL samples:

  • mqtts://mqtt.eclipse.org: MQTT over SSL, port 8883
  • mqtts://mqtt.eclipse.org:8884: MQTT over SSL, port 8884

• MQTT over Websocket samples:

  • ws://mqtt.eclipse.org:80/mqtt

• MQTT over Websocket Secure samples:

  • wss://mqtt.eclipse.org:443/mqtt

• 最小配置:

  const esp_mqtt_client_config_t mqtt_cfg = {
      .uri = "mqtt://mqtt.eclipse.org",
      // .user_context = (void *)your_context
  };
  esp_mqtt_client_handle_t client = esp_mqtt_client_init(&mqtt_cfg);
  
  //注册回调函数
  esp_mqtt_client_register_event(client, ESP_EVENT_ANY_ID, mqtt_event_handler, client);
  esp_mqtt_client_start(client);
  

• 注意，默认mqtt客户端使用事件句柄来处理mqtt的事件，如连接，订阅，发布等等。

3.2 SSL

3.3 遗嘱

MQTT支持使用遗嘱消息，当客户端意外断开连接时，遗嘱消息被服务端发送并用于通知其他客户端。在 esp_mqtt_client_config_t 中可配置遗嘱消息：

• lwt_topic ：指向遗嘱消息的主题
• lwt_msg：指向遗嘱消息
• lwt_msg_len：消息有效载荷长度
• lwt_qos：消息服务质量
• lwt_retain：是否保留

3.4 其他配置参数

• disable_clean_session：默认清除会话，对于连接消息，该参数关闭清除会话标志
• keepalive：保活时间，默认120s
• disable_auto_reconnect：关闭自动重连
• user_context：本地参数，用于传递到事件处理句柄
• task_prio：mqtt任务等级，默认5
• task_stack：mqtt堆栈默认6144 bytes，menuconfig可配置
• buffer_size：接收和缓存的长度，默认1024 bytes
• username：连接到broker的用户名（服务器）
• password：连接到broker的密码（服务器）
• client_id：指向客户端id，一般是一个唯一的字符串，可从broker中获取
• host：MQTT broker的ip地址，域名。设置了url会重写该参数
• port：MQTT broker的端口。设置了url会重写该参数
• transport：设置传输协议。设置了url会重写该参数
• refresh_connection_after_ms：在多少时间（ms）后刷新连接
• event_handle：处理mqtt事件的回调函数
• event_loop_handle：mqtt事件组库

更多esp_mqtt_client_config_t的选项，请参考下面API：

3.5 项目配置菜单来配置mqtt

通过idf.py menuconfig项目配置菜单来配置mqtt，在Component config -> ESP-MQTT Configuration

下面的设置是可行的：

• CONFIG_MQTT_PROTOCOL_311：使用3.1.1版本的MQTT协议
• CONFIG_MQTT_TRANSPORT_SSL, CONFIG_MQTT_TRANSPORT_WEBSOCKET：选择传输层协议
• CONFIG_MQTT_CUSTOM_OUTBOX：使用本地邮箱

3.6 事件

mqtt主要围绕以下事件进行数据的处理：

• MQTT_EVENT_BEFORE_CONNECT：客户端初始化完成并开始连接到broker
• MQTT_EVENT_CONNECTED：客户端成功与broker建立连接，客户端准备好接收发送数据
• MQTT_EVENT_DISCONNECTED：客户端由于无法接收或者发送消息而断开连接
• MQTT_EVENT_SUBSCRIBED：broker 确认客户端的订阅请求。保留订阅消息的id
• MQTT_EVENT_UNSUBSCRIBED：broker确认了客户端的取消订阅消息。保留取消订阅消息的id
• MQTT_EVENT_PUBLISHED：broker确认了用户发布的消息。仅对qos为1和2的消息有效，保留发布消息的id
• MQTT_EVENT_DATA：客户端已接收到一个发布的消息。event data包括：消息id，主题名称，数据及数据长度。若数据长度超过buffer大小，则多个MQTT_EVENT_DATA事件会被触发，*current_data_offset和*total_data_len用于保持对数据的追踪。在此可将数据读到缓存中。
• MQTT_EVENT_ERROR：客户端遇到错误。esp_mqtt_error_type_t from error_handle in the event data 可用于判断是哪个类型的错误。

四，API参考

esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *config)

根据配置创建mqtt 客户端句柄。

esp_err_t esp_mqtt_client_set_uri(esp_mqtt_client_handle_t client, const char *uri);

设置mqtt连接的url，这个函数通常会重写esp_mqtt_client_init里的配置。

esp_err_t esp_mqtt_client_start(esp_mqtt_client_handle_t client);

开启mqtt客户端

esp_err_t esp_mqtt_client_reconnect(esp_mqtt_client_handle_t client);

用于强制重新连接

esp_err_t esp_mqtt_client_disconnect(esp_mqtt_client_handle_t client);

用于强制从broker中断开连接

esp_err_t esp_mqtt_client_stop(esp_mqtt_client_handle_t client);

停止mqtt客户端任务，不能再event handle中调用。

int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic, int qos);

客户端订阅指向服务质量的主题。

• 客户端必须已经连接，该API可被用户任务或mqtt event 回调函数调用。该API使用信号量，所以可能导致一段时间的阻塞。

• 返回：消息id

int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *topic);

取消订阅指定主题

• 客户端必须连接
• 线程安全。参考上面
• 返回：消息id

int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic, const char *data, int len, int qos, int retain);

客户端发布消息到broker

• 该API可能会阻塞几秒钟（由于网络，数据长度问题）
• 客户端不必连接
• 线程安全
• 返回：消息id

参数：

• client：客户端句柄
• topic：主题字符串
• data：有效载荷字符串
• len：有效载荷字符串长度
• qos：消息服务质量
• retain：保留消息标志

esp_err_t esp_mqtt_client_destroy(esp_mqtt_client_handle_t client);

销毁mqtt客户端，不能再mqtt回调函数中调用

esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_client_config_t *config);

设置配置结构体，通常用于更新mqtt配置，再连接之前的回调事件中。

esp_err_t esp_mqtt_client_register_event(esp_mqtt_client_handle_t client, esp_mqtt_event_id_t event, esp_event_handler_t event_handler, void* event_handler_arg);

注册mqtt事件

参数：

• client：

• event：事件类型

• event_handler:事件处理回调函数

• event_handler_arg:事件处理回调函数传入参数

int esp_mqtt_client_get_outbox_size(esp_mqtt_client_handle_t client);

获取邮箱大小

六，示例

 

在上一篇文章中，说明了MQTT在树莓派部署服务器的方法，和手机客户端作为SUB端的方法。

本节终于到了端PUB的方法，毕竟物联网的物就在这个地方。

ESP32网上介绍有两个库，一个subpubclient和MQTTclient两个，subpub试了一下，较复杂，需要另外使用wifi.h的库，集成度较低，适合深度开发。

而MQTTclient高集成度，只需要实例化一个EspMQTTClient对象

EspMQTTClient client(
  "WifiSSID",
  "WifiPassword",
  "192.168.1.100",  // MQTT Broker server ip
  "MQTTUsername",   // Can be omitted if not needed
  "MQTTPassword",   // Can be omitted if not needed
  "TestClient",     // Client name that uniquely identify your device
  1883              // The MQTT port, default to 1883. this line can be omitted
);

依次填入wifi名字，密码，MQTT toker的地址，MQTT的用户名，密码，自己给自己PUB起一个名字。

---

在setup中保持例子的设置，大概是保持串口输出等

  // Optionnal functionnalities of EspMQTTClient : 
  client.enableDebuggingMessages(); // Enable debugging messages sent to serial output
  client.enableHTTPWebUpdater(); // Enable the web updater. User and password default to values of MQTTUsername and MQTTPassword. These can be overrited with enableHTTPWebUpdater("user", "password").
  client.enableLastWillMessage("TestClient/lastwill", "I am going offline"); 

接下来就是在loop中使用

  client.loop();

使PUB端一直运行。

---

好，这不是什么没发吗？答案在

void onConnectionEstablished()

函数中，这个函数是loop内建调用函数，相当于中断执行函数，loop里面只是一个状态机而已。

在这个函数中，使用

client.publish("mytopic/test", "This is a message");

来实现发布，第一个参数为topic，第二个参数为内容，so easy，美滋滋。

用

  client.subscribe("mytopic/test", [](const String & payload) {
    Serial.println(payload);
  });

来实现订阅到串口打印，so so easy是不是，站在巨人肩上就是不一样。

---

在

void onConnectionEstablished()

可以加一个while来循环发送消息。忘加delay函数了，以为

client.executeDelayed(5 * 1000, []()

可以有延时作用，但是while在这个delay后，直接又回到第一条执行，结果手机上订阅的消息瞬间炸了。。。手机app也死了。。。

以为树莓派也死了，结果好好的，说明toker的作用，转发而已，跟http服务器有点不一样。

加个delay效果很好。交替发送。

如果不深入修改的话推荐这个库，很好上手，免去wifi+MQTT繁琐的设置，我重启esp32后重连wifi也很快，退出机制也比较完善。

---

终于在最后5天前基本完成MQTT的小目标，GOOD。

想要搭建MQTT服务器，请参考

☞☞ 搭建MQTT服务器篇

服务器平台：Ubuntu20.04 + mosquitto
硬件平台：ESP32开发板
MQTT客户端：mosquitto、 通信猫、 mqttfx等等
开发平台：Thoony 3.2

先看下 microPython 支持哪些模块吧:

>>> help('modules')
__main__          gc                ubinascii         upysh
_boot             inisetup          ubluetooth        urandom
_onewire          machine           ucollections      ure
_thread           math              ucryptolib        urequests
_uasyncio         micropython       uctypes           uselect
_webrepl          neopixel          uerrno            usocket
apa106            network           uhashlib          ussl
btree             ntptime           uhashlib          ustruct
builtins          onewire           uheapq            usys
cmath             uarray            uio               utime
dht               uasyncio/__init__ ujson             utimeq
ds18x20           uasyncio/core     umqtt/robust      uwebsocket
esp               uasyncio/event    umqtt/simple      uzlib
esp32             uasyncio/funcs    uos               webrepl
flashbdev         uasyncio/lock     upip              webrepl_setup
framebuf          uasyncio/stream   upip_utarfile     websocket_helper
Plus any modules on the filesystem

我们这里将用到 umqtt/simple模块中的mqtt客户端、 machine模块中的Pin、network模块、time模块。

这里主要演示如何远程点led灯和监控led灯的状态。

用microPython写l简单的测试demo程序

from umqtt.simple import MQTTClient
from machine import Pin
import network
import time

ssid='your ssid'
passwd='your passwd'
client_id = "slim_id"
mserver = '192.168.0.135'
port=1883

topic_ctl = b'led_ctl'#设备订阅的主题,客户端推送消息的主题
topic_sta = b'led_sta'#客户端订阅的主题,设备推送消息的主题
client = None
wlan = None

led1=Pin(2, Pin.OUT, value=1)
led2=Pin(5, Pin.OUT, value=1)
    
def sub_callback(topic, msg):
    """
    收到订阅消息回调
    """
    global client
    print((topic_ctl, msg))
    if msg == b'led1 ON' or msg == b'led1 on':
        pub_msg = 'LED1: ON-state'
        led1.value(0)
    elif msg == b'led1 OFF' or msg == b'led1 off':
        pub_msg = 'LED1: OFF-state'
        led1.value(1)
    elif msg == b'led2 ON' or msg == b'led2 on':
        pub_msg = 'LED2: ON-state'
        led2.value(0)
    elif msg == b'led2 OFF' or msg == b'led2 off':
        pub_msg = 'LED2: OFF-state'
        led2.value(1)
    else:
        pub_msg = 'other msg'
    client.publish(topic_sta, pub_msg, retain=True)

def connectWiFi():
    """
    连接到WiFi网络
    """
    global wlan
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    wlan.disconnect()
    wlan.connect(ssid, passwd)
    while(wlan.ifconfig()[0]=='0.0.0.0'):
        time.sleep(1)

try:
    connectWiFi()
    client = MQTTClient(client_id, mserver, 0)
    client.set_callback(sub_callback)
    client.connect()
    client.subscribe(topic_ctl)
    client.publish(topic_sta, 'ESP32 Device online', retain=True)
    print("Connected to %s, subscribed to %s topic" % (mserver, topic_ctl))
    while True:
        client.wait_msg()
finally:
    if client is not None:
        print('off line')
        client.disconnect()
    wlan.disconnect()
    wlan.active(False)

测试

通信猫测试端

订阅： led_sta
发布： led_ctl

Thonny ESP32端

设备终端
订阅：led_ctl
发布：led_sta

手机APP端

服务器端

mosquitto mqtt服务器
订阅： led_sta 、 led_ctl

一个物联网属性的LED灯就做好了