乐趣区

关于物联网:如何在-Flask-项目中使用-MQTT

jiezi

Flask 是一个应用 Python 编写的轻量级 Web 利用框架，其被称为“微框架”，因为它应用简略的外围，用扩大减少其余性能，例如：ORM、窗体验证工具、文件上传、各种开放式身份验证技术等。

MQTT 是一种基于公布 / 订阅模式的 轻量级物联网音讯传输协定，能够用极少的代码和带宽为联网设施提供实时牢靠的音讯服务，它广泛应用于物联网、挪动互联网、智能硬件、车联网、电力能源等行业。

本文次要介绍如何在 Flask 我的项目中实现 MQTT 客户端与 MQTT 服务器的连贯、订阅、勾销订阅、收发音讯等性能。

咱们将应用到 Flask-MQTT 客户端库，它是一个 Flask 扩大，能够看作一个 paho-mqtt 的装璜器，用于简化 Flask 应用程序中的 MQTT 集成。

我的项目初始化

本我的项目应用 Python 3.8 进行开发测试，读者可用如下命令确认 Python 的版本。

$ python3 --version
Python 3.8.2

应用 Pip 装置 Flask-MQTT 库 

pip3 install flask-mqtt

Flask-MQTT 应用

本文将应用 EMQ 提供的收费公共 MQTT 服务器，该服务基于 MQTT 云服务 – EMQX Cloud 创立。服务器接入信息如下：

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

导入 Flask-MQTT

导入 Flask 库以及 Flask-MQTT 扩大，并创立 Flask 利用 

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

app = Flask(__name__)

配置 Flask-MQTT 扩大 

app.config['MQTT_BROKER_URL'] = 'broker.emqx.io'
app.config['MQTT_BROKER_PORT'] = 1883
app.config['MQTT_USERNAME'] = ''  # 当你须要验证用户名和明码时，请设置该项
app.config['MQTT_PASSWORD'] = ''  # 当你须要验证用户名和明码时，请设置该项
app.config['MQTT_KEEPALIVE'] = 5  # 设置心跳工夫，单位为秒
app.config['MQTT_TLS_ENABLED'] = False  # 如果你的服务器反对 TLS，请设置为 True
topic = '/flask/mqtt'

mqtt_client = Mqtt(app)

残缺的配置项能够参考 Flask-MQTT 配置文档。

编写连贯回调函数

能够在该回调函数中对 MQTT 连贯胜利或失败的状况进行解决，本示例将在连贯胜利后订阅 /flask/mqtt 主题。

@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)

编写音讯回调函数

该函数将打印 /flask/mqtt 主题接管到的音讯。

@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))

创立公布音讯接口

咱们创立一个简略的 POST 接口实现 MQTT 音讯公布。

在理论利用中该接口可能须要进行一些更简单的业务逻辑解决。

@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]})

运行 Flask 利用 

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

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

测试

接下来咱们应用 MQTT 客户端 – MQTT X 进行连贯、订阅、公布测试。

测试音讯接管

  1. 在 MQTT X 中创立链接并连贯到服务器。

  2. 在 MQTT X 中向 /flask/mqtt 主题公布音讯 Hello from MQTT X。

  3. 在 Flask 运行窗口中将能看到 MQTT X 发送的音讯。

测试音讯公布接口

  1. 在 MQTT X 中订阅 /flask/mqtt 主题。

  2. 应用 Postman 调用 /publish 接口：发送音讯 Hello from Flask 至 /flask/mqtt 主题。

  3. 在 MQTT X 中将能看到 Flask 发送过去的音讯。

残缺代码 

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'] = ''  # 当你须要验证用户名和明码时，请设置该项
app.config['MQTT_PASSWORD'] = ''  # 当你须要验证用户名和明码时，请设置该项
app.config['MQTT_KEEPALIVE'] = 5  # 设置心跳工夫，单位为秒
app.config['MQTT_TLS_ENABLED'] = False  # 如果你的服务器反对 TLS，请设置为 True
topic = '/flask/mqtt'

mqtt_client = Mqtt(app)


@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(host='127.0.0.1', port=5000)

注意事项

Flask-MQTT 目前不适宜应用多个工作实例，如果您须要应用 gevent 或 gunicorn 这样的 WSGI 服务器，请确保只有一个工作实例。

总结

至此，咱们应用 Flask-MQTT 实现了简略的 MQTT 客户端，并且能够在 Flask 利用中订阅、公布音讯。

版权申明：本文为 EMQ 原创，转载请注明出处。

原文链接：https://www.emqx.com/zh/blog/how-to-use-mqtt-in-flask

退出移动版