【小沐学Python】Python实现通信协议（mqtt）

1、简介

https://pypi.org/project/paho-mqtt/

MQTT 协议是一种机器对机器 （M2M）/“物联网”连接协议。它被设计为一种极其轻量级的发布/订阅消息传递传输，对于需要少量代码占用空间和/或网络带宽非常宝贵的远程位置的连接非常有用。

MQTT是一种轻量级的发布/订阅模式的通信协议，特别适用于低带宽、高延迟或不稳定的网络环境。正因为其高效和低耗，MQTT已成为物联网通信的标准。

MQTT是一种标准的二进制发布-订阅消息传递协议，专为快速可靠地传输工业资产、系统和应用程序数据而设计，以实现预测性维护，尤其适用于非常受限的条件下。限制可能包括不可靠的网络连接、有限的带宽或有限的电池电量。MQTT建立在TCP/IP之上，这是互联网上连接网络设备的首选通信协议。因此，MQTT非常适合IIoT，支持事件驱动架构。

MQTT技术旨在将数据推送至企业内数千个远程资产、系统和应用程序，并从中获取数据。MQTT Sparkplug是一个位于MQTT之上的框架，为工业数据添加更多上下文。它是一个开源软件规范，为MQTT客户端提供了一个框架，以集成各种工业数据，并通过定义数据模型提供上下文。它为制造设备制造商和软件提供商提供了一种一致的方式来共享具有上下文的工厂数据，丰富了预测性维护数据。

MQTT允许对车辆进行实时追踪，提供最新的位置和状态更新。这项能力有助于监控和改进交付时间表，优化路线，并提高整体车队效率。

MQTT最大的优势之一是能够具有统一的命名空间（UNS），作为来自不同系统（如工厂机器、质量系统、MES、ERP等）的所有OEE数据的单一真实来源，然后提供给可以使用和行动的应用程序。（支持不同数据类型和数据速率）

2、安装

pip install paho-mqtt
# pip3 install -i https://pypi.doubanio.com/simple paho-mqtt

# or
git clone https://github.com/eclipse/paho.mqtt.python
cd paho.mqtt.python
python setup.py install

3、基本代码

3.1 基于Apollo服务

• 服务器端搭建：
  https://archive.apache.org/dist/activemq/activemq-apollo/1.7.1/
• 解压并打开目标如下：
  pollo中间件其实是免安装的，我们只需要下载apache-apollo-1.7.1-windows-distro.zip,然后解压到某个文件夹就可以了。在这里我解压到D:\soft\apache-apollo-1.7.1。
  
  （1）先到apollo目录下 cd D:\soft\apache-apollo-1.7.1
  （2）D:\soft\apache-apollo-1.7.1\bin
  （3）执行命令apollo create myapollo

出现如下错误 Loading configuration file ‘D:\phpStudy\apache-apollo-1.7.1\bin\mybroker\etc\apollo.xml’.
Startup failed: java.lang.NoClassDefFoundError: javax/xml/bind/ValidationEventHandler
答：换上jdk1.8版本即可。

apollo-broker.cmd run

进入后台管理，打开网页，输入ip + : 61680 进入后台管理 ，默认用户名admin 密码 password，例如 127.0.0.1:61680

http://127.0.0.1:61680

• 订阅主题

import time

import paho.mqtt.client as mqtt
# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("连接成功")
        print("Connected with result code " + str(rc))

def on_message(client, userdata, msg):
    print(msg.topic + " " + str(msg.payload))
client = mqtt.Client(protocol=3)
client.username_pw_set("admin", "password")
client.on_connect = on_connect
client.on_message = on_message
client.connect(host="127.0.0.1", port = 61613, keepalive=60)  # 订阅频道
time.sleep(1)
# client.subscribe("public")
client.subscribe([("public", 0), ("test", 2)])
client.loop_forever()

• 发布主题

import paho.mqtt.client as mqtt
import time
import sys
def on_connect(client, userdata, flags, rc):
    print("Connected with result code " + str(rc))
def on_subscribe(client,userdata,mid,granted_qos):
    print("消息发送成功")
client = mqtt.Client(protocol=3)
client.username_pw_set("admin", "password")
client.on_connect = on_connect
client.on_subscribe = on_subscribe
client.connect(host="127.0.0.1", port = 61613, keepalive=60)  # 订阅频道
time.sleep(1)
i = 0
while True:
    try:
        # 发布MQTT信息
        sensor_data = "test" + str(i)
        client.publish(topic="public", payload=sensor_data, qos=0)
        time.sleep(5)
        i += 1
    except KeyboardInterrupt:
        print("EXIT")
        client.disconnect()
        sys.exit(0)

执行以上脚本之后：

3.2 基于公共服务

在此，我们选择使用EMQX提供的免费MQTT公共服务器，但同样可以选择其他任何MQTT broker。

Broker: broker.emqx.io
TCP Port: 1883
Websocket Port: 8083

• 消息发布代码，pub.py

# python 3.6
 
import random
import time
 
from paho.mqtt import client as mqtt_client
 
 
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 1000)}'
 
 
def connect_mqtt():
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)
 
    client = mqtt_client.Client(client_id)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client
 
 
def publish(client):
    msg_count = 0
    while True:
        time.sleep(1)
        msg = f"messages: {msg_count}"
        result = client.publish(topic, msg)
        # result: [0, 1]
        status = result[0]
        if status == 0:
            print(f"Send `{msg}` to topic `{topic}`")
        else:
            print(f"Failed to send message to topic {topic}")
        msg_count += 1
 
 
def run():
    client = connect_mqtt()
    client.loop_start()
    publish(client)
 
 
if __name__ == '__main__':
    run()

• 消息订阅代码，sub.py

# python3.6
 
import random
 
from paho.mqtt import client as mqtt_client
 
 
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 100)}'
 
 
def connect_mqtt() -> mqtt_client:
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)
 
    client = mqtt_client.Client(client_id)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client
 
 
def subscribe(client: mqtt_client):
    def on_message(client, userdata, msg):
        print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")
 
    client.subscribe(topic)
    client.on_message = on_message
 
 
def run():
    client = connect_mqtt()
    subscribe(client)
    client.loop_forever()
 
 
if __name__ == '__main__':
    run()

4、更多代码（flask）

4.1 安装flask_mqtt库

pip install flask_mqtt

4.2 编写flask_mqtt脚本

编写代码：

from flask import Flask, request, jsonify
from flask_mqtt import Mqtt

app = Flask(__name__)

# 代理地址
app.config['MQTT_BROKER_URL'] = 'broker.emqx.io'
# 端口
app.config['MQTT_BROKER_PORT'] = 1883
# 当需要验证用户名和密码时，请设置该项
app.config['MQTT_USERNAME'] = 'user'
# 当需要验证用户名和密码时，请设置该项
app.config['MQTT_PASSWORD'] = '123456'
# 设置心跳时间，单位为秒
app.config['MQTT_KEEPALIVE'] = 60
# 如果服务器支持 TLS，则设置为 True
app.config['MQTT_TLS_ENABLED'] = False
# 主题
topic = '/flask/mqtt'
# 实例化
mqtt_client = Mqtt(app)

@app.route('/')
def index():
    # 初始路由
    return "Welcome mqtt_flask"

@mqtt_client.on_connect()
def handle_connect(client, userdata, flags, rc):
    """连接回调函数"""
    if rc == 0:
        print('Connected successfully')
        # 订阅主题
        mqtt_client.subscribe(topic)
    else:
        # 连接失败
        print('Bad connection. Code:', rc)

@mqtt_client.on_message()
def handle_mqtt_message(client, userdata, message):
    """ 消息回调函数 """
    # 定义接受到的消息
    data = dict(
        # 主题
        topic=message.topic,
        # 内容
        payload=message.payload.decode()
    )
    # 打印输出接收到的消息
    print('Received message on topic: {topic} with payload: {payload}'.format(**data))

@app.route('/publish', methods=['POST'])
def publish_message():
    """ 消息发布接口（实际应用中，该接口可能需要处理一些复杂业务逻辑） """
    # 格式化数据
    request_data = request.get_json()
    # 发布消息
    publish_result = mqtt_client.publish(request_data['topic'], request_data['msg'])

    return jsonify({'code': publish_result[0]})

if __name__ == '__main__':
    # app.run()
    app.run(host='127.0.0.1', port=5000)

当 Flask 应用启动后，MQTT 客户端将会连接到服务器，并且订阅主题 /flask/mqtt。

4.3 安装MQTTX（MQTT 客户端）

https://mqttx.app/zh/downloads

主界面显示如下：

4.4 测试消息接收

• 创建连接

Host：为代码中定义好的 broker.emqx.io
Port：为代码中定义好的 1883
用户名、密码根据需要添加

• 添加订阅
  主题为：/flask/mqtt
  
  
• 在MQTTX中发布消息
  
  Flask控制台中接收到的消息：
  

4.5 测试消息发布

• 订阅使用消息接收的订阅
  主题为：/flask/mqtt

使用 Postman 调用 /publish 接口：发送消息 Hello from Flask 至 /flask/mqtt 主题。

在 MQTTX 中将能看到 Flask 发送过来的消息。

在 python中将能看到 Flask 发送过来的消息。

结语

如果您觉得该方法或代码有一点点用处，可以给作者点个赞，或打赏杯咖啡；╮(￣▽￣)╭
如果您感觉方法或代码不咋地//(ㄒoㄒ)//，就在评论处留言，作者继续改进；o_O???
如果您需要相关功能的代码定制化开发，可以留言私信作者；(????)
感谢各位大佬童鞋们的支持！( ′ ▽′ )ﾉ ( ′ ▽′)っ！！！