赞
踩
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数据的单一真实来源,然后提供给可以使用和行动的应用程序。(支持不同数据类型和数据速率)
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
出现如下错误 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)
执行以上脚本之后:
在此,我们选择使用EMQX提供的免费MQTT公共服务器,但同样可以选择其他任何MQTT broker。
Broker: broker.emqx.io
TCP Port: 1883
Websocket Port: 8083
# 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()
# 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()
pip install 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。
https://mqttx.app/zh/downloads
主界面显示如下:
Host:为代码中定义好的 broker.emqx.io
Port:为代码中定义好的 1883
用户名、密码根据需要添加
使用 Postman 调用 /publish 接口:发送消息 Hello from Flask 至 /flask/mqtt 主题。
在 MQTTX 中将能看到 Flask 发送过来的消息。
在 python中将能看到 Flask 发送过来的消息。
如果您觉得该方法或代码有一点点用处,可以给作者点个赞,或打赏杯咖啡;
╮( ̄▽ ̄)╭
如果您感觉方法或代码不咋地
//(ㄒoㄒ)//,就在评论处留言,作者继续改进;
o_O???
如果您需要相关功能的代码定制化开发,可以留言私信作者;
(✿◡‿◡)
感谢各位大佬童鞋们的支持!
( ´ ▽´ )ノ ( ´ ▽´)っ!!!
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。