[Python]MQTT介绍与使用

[Python] MQTT介绍与使用

MQTT协议

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

特点

• 开放消息协议&＃xff0c;简单易实现
• 发布订阅模式&＃xff0c;一对多消息发布
• 基于TCP/IP网络连接,提供有序&＃xff0c;无损&＃xff0c;双向连接。
• 1字节固定报头&＃xff0c;2字节心跳报文&＃xff0c;最小化传输开销和协议交换&＃xff0c;有效减少网络流量。
• 消息QoS支持&＃xff0c;可靠传输保证

原理

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

MQTT传输的消息分为&＃xff1a;主题&＃xff08;Topic&＃xff09;和负载&＃xff08;payload&＃xff09;两部分&＃xff1a;

• &＃xff08;1&＃xff09;Topic&＃xff0c;可以理解为消息的类型&＃xff0c;订阅者订阅&＃xff08;Subscribe&＃xff09;后&＃xff0c;就会收到该主题的消息内容&＃xff08;payload&＃xff09;

• &＃xff08;2&＃xff09;payload&＃xff0c;可以理解为消息的内容&＃xff0c;是指订阅者具体要使用的内容。

MQTT代理

mosquitto是一款开源的MQTT消息代理&＃xff08;服务器&＃xff09;软件&＃xff0c;提供轻量级的&＃xff0c;支持可发布/可订阅的的消息推送模式&＃xff0c;使设备对设备之间的短消息通信变得简单&＃xff0c;比如现在应用广泛的低功耗传感器&＃xff0c;手机、嵌入式计算机、微型控制器等移动设备。

• 安装mosquitto

sudo apt-get install mosquitto -y


• 安装mosquitto命令行客户端

sudo apt-get install mosquitto-clients -y


使用

发布使用mosquitto_pub命令,订阅使用mosquitto_sub命令,常用参数如下

发布

mosquitto_sub -h localhost -t "test/#" -u user2 -P 123456 -i "client1"


订阅

mosquitto_pub -h localhost -t "test/abc" -u user1 -P 123456 -i "client3" -m "How are you?"


MQTT使用

MQTT 客户端工具

MQTT 客户端工具常用于建立与 MQTT 服务器 的连接&＃xff0c;进行主题订阅、消息收发等操作

• MQTT X 是由 EMQ 开源的一款跨平台 MQTT 5.0 桌面测试客户端&＃xff0c;支持 macOS&＃xff0c;Linux&＃xff0c;Windows

  

  项目地址&＃xff1a; MQTT X 官网

  下载地址&＃xff1a; MQTT X GitHub

• mqtt-spy是 Eclipse Paho 和 Eclipse IoT 的一部分&＃xff0c;它通过直接启动 JAR 文件在 Java 8 和 JavaFX 之上运行&＃xff0c;mqtt-spy 有一种很好的交互方式来展现基本的 MQTT发布/订阅机制。

  项目地址&＃xff1a; GitHub mqtt-spy

  下载地址&＃xff1a; https://github.com/eclipse/paho.mqtt-spy/releases

• MQTT Explorer 是一个全面且易于使用的 MQTT 客户端&＃xff0c;是目前比较流行的 MQTT 桌面测试客户端之一&＃xff0c;基于它提供有关 MQTT Topics 的结构化预览展示&＃xff0c;并使其在对 MQTT Broker 上的设备/服务的使用变得非常简单。目前基于 CC BY-NC-ND 4.0 协议开源&＃xff0c;用户可随意查看源码和使用。

  

  项目地址&＃xff1a; Github MQTT-Explorer

  下载地址&＃xff1a; https://mqtt-explorer.com/

Python

使用 paho-mqtt 库 &＃xff0c;实现客户端与 MQTT 服务器的连接、订阅、取消订阅、收发消息等功能

• 使用connect()/connect_async() 连接MQTT代理
• 频繁的调用loop()来维持与MQTT代理之间的流量或者使用loop_start()来设置一个线程为你调用loop()或者在一个阻塞的函数中调用loop_forever()来为你调用loop()
• 使用subscribe()订阅一个主题&＃xff08;topic&＃xff09;并接受消息&＃xff08;messages&＃xff09;
• 使用publish()来发送消息
• 使用disconnect()来断开与MQTT代理的连接

详细见paho-mqtt项目描述

回调 Callbacks

• on_connect 当代理响应连接请求时调用
• on_disconnect 当与代理断开连接时调用
• on_message 当收到关于客户订阅的主题的消息时调用。
• on_publish 当使用publish()发送的消息已经传输到代理时被调用
• on_subscribe 当代理响应订阅请求时被调用。
• on_unsubscribe 当代理响应取消订阅请求时调用
• on_log 当客户端有日志信息时调用

Client

• 构造函数
  Client(client_id&＃61;"", clean_session&＃61;True, userdata&＃61;None, protocol&＃61;MQTTv311, transport&＃61;"tcp")

  参数	含义
  client_id	连接到代理时使用的唯一客户端ID字符串。 如果client_id长度为零或无&＃xff0c;则会随机生成一个。 在这种情况下&＃xff0c;clean_session参数必须为True。
  clean_session	一个决定客户端类型的布尔值。 如果为True&＃xff0c;那么代理将在其断开连接时删除有关此客户端的所有信息。 如果为False&＃xff0c;则客户端是持久客户端&＃xff0c;当客户端断开连接时&＃xff0c;订阅信息和排队消息将被保留。
  userdata	用户定义的任何类型的数据作为userdata参数传递给回调函数。 它可能会在稍后使用user_data_set&＃xff08;&＃xff09;函数进行更新。
  protocol	用于此客户端的MQTT协议的版本。 可以是MQTTv31或MQTTv311或MQTT v5.0。
  transport	设置为“websockets”通过WebSockets发送MQTT。 保留默认的“tcp”使用原始TCP。

• reinitialise

  reinitialise(client_id&＃61;"", clean_session&＃61;True, userdata&＃61;None)

  reinitialise&＃xff08;&＃xff09;函数将客户端重置为其开始状态&＃xff0c;就像它刚刚创建一样。 它采用与Client&＃xff08;&＃xff09;构造函数相同的参数。

• connect
  将客户端连接到代理。 这是一个阻塞函数。

  connect(host, port&＃61;1883, keepalive&＃61;60, bind_address&＃61;"")

  参数	含义
  host	远程代理的主机名或IP地址
  port	要连接的服务器主机的网络端口。 默认为1883
  keepalive	与代理通信之间允许的最长时间段&＃xff08;以秒为单位&＃xff09;。 如果没有其他消息正在交换&＃xff0c;则它将控制客户端向代理发送ping消息的速率
  bind_address	假设存在多个接口&＃xff0c;将绑定此客户端的本地网络接口的IP地址

Publish

从客户端发送消息给代理
publish(topic, payload&＃61;None, qos&＃61;0, retain&＃61;False)

消息将会发送给代理&＃xff0c;并随后从代理发送到订阅匹配主题的任何客户端

返回以下属性和方法的MQTTMessageInfo
rc:发布的结果

mid:发布请求的消息ID。
如果mid已定义&＃xff0c;则可以通过检查on_publish&＃xff08;&＃xff09;回调中的mid来跟踪发布请求。
wait_for_publish():函数将阻塞&＃xff0c;直到消息发布。 如果消息未排队&＃xff08;rc &＃61;&＃61; MQTT_ERR_QUEUE_SIZE&＃xff09;&＃xff0c;它将引发ValueError。
is_published:如果消息已发布&＃xff0c;is_published返回True。 如果消息未排队&＃xff08;rc &＃61;&＃61; MQTT_ERR_QUEUE_SIZE&＃xff09;&＃xff0c;它将引发ValueError。

如果主题为无&＃xff0c;长度为零或无效&＃xff08;包含通配符&＃xff09;&＃xff0c;qos不是0,1或2之一&＃xff0c;或者有效负载长度大于268435455字节&＃xff0c;则会引发ValueError。

Subscribe

订阅一个或多个主题

subscribe(topic, qos&＃61;0

Example

#!/usr/bin/env python3
# -*- coding: utf-8 -*-import paho.mqtt.client as mqtt
import jsonCONN_CONFIG &＃61; {&＃39;product&＃39;: {&＃39;host&＃39;: &＃39;mqtt.broker.com&＃39;,&＃39;port&＃39;: 1883,&＃39;username&＃39;: &＃39;username&＃39;,&＃39;password&＃39;: &＃39;password&＃39;},&＃39;dev&＃39;: {&＃39;host&＃39;: &＃39;127.0.0.1&＃39;,&＃39;port&＃39;: 1883,&＃39;username&＃39;: &＃39;username&＃39;,&＃39;password&＃39;: &＃39;password&＃39;}
}class MQTTClient:def __init__(self, client_id, profile&＃61;&＃39;dev&＃39;, ):self.config &＃61; CONN_CONFIG[profile]self.client_id &＃61; client_idself.connect_flag &＃61; Falseself.client &＃61; self.__create_client(client_id)self.__connect()def __create_client(self, client_id):client &＃61; mqtt.Client(client_id&＃61;client_id)client.username_pw_set(self.config[&＃39;username&＃39;], self.config[&＃39;password&＃39;])client.on_connect &＃61; self.__on_connectclient.on_message &＃61; self.__on_messageclient.on_publish &＃61; self.__on_publishclient.on_disconnect &＃61; self.__on_disconnectreturn clientdef __connect(self):self.client.connect(self.config[&＃39;host&＃39;], self.config[&＃39;port&＃39;], 60)self.client.loop_start()def publish(self, msg):if not self.connect_flag:self.__connect()payload &＃61; dict(data&＃61;msg)self.client.publish(f&＃39;topic&＃39;, payload&＃61;json.dumps(payload), qos&＃61;0)def __on_connect(self, client, userdata, flags, rc):self.connect_flag &＃61; Trueprint("connected with result code: " &＃43; str(rc))def __on_message(self, client, userdata, msg):print(msg.topic &＃43; " " &＃43; str(msg.payload))def __on_subscribe(self, client, userdata, mid, granted_qos):print(&＃39;on subscribe: mid&＃61;&＃39; &＃43; str(mid))def __on_publish(self, client, userdata, mid):print("on publish: mid&＃61;" &＃43; str(mid))def __on_disconnect(self, client, userdata, rc):self.connect_flag &＃61; Falseprint(&＃39;disconnecting reason &＃39; &＃43; str(rc))def __del__(self):self.client.disconnect()

参考

MQTT 入门介绍
MQTT中文网
ubuntu下Mosquitto安装及配置
常见MQTT 客户端工具比较
Python paho-mqtt 模块使用和API分析