MQTT服务搭建及python使用示例

1、MQTT协议

1.1、MQTT介绍

        MQTT（Message Queuing Telemetry Transport）是一种轻量级的、基于发布/订阅模式的通信协议，通常用于物联网设备之间的通讯。它具有低带宽、低功耗和开放性等特点，适合在网络带宽有限或者网络连接不稳定的环境下使用。MQTT协议使用TCP/IP协议栈进行通讯，支持多种编程语言和平台，并且能够提供可靠的消息传递机制。

        在MQTT中，设备可以发布消息到特定的主题（topic），同时其他设备可以订阅这些主题以接收相应的消息。这种发布/订阅模式使得设备之间的通讯更加灵活和高效。MQTT还支持三种服务质量等级：至多一次（at most once）、至少一次（at least once）和恰好一次（exactly once），以满足不同场景下对消息传递可靠性的需求。

        优点：代码量少，开销低，带宽占用小，即时通讯协议。

1.2、MQTT概念

        订阅(Subscribtion)：订阅包含主题筛选器（ Topic Filter ）和最大服务质量（ QoS ）。订阅会与一个会话（ Session ）关联。一个会话可以包含多个订阅。每一个会话中的每个订阅都有一个不同的主题筛选器。

        会话（Session）：每个客户端与服务器建立连接后就是一个会话，客户端和服务器之间有状态交互。会话存在于一个网络之间，也可能在客户端和服务器之间跨越多个连续的网络连接。

        主题名（Topic Name）：连接到一个应用程序消息的标签，该标签与服务器的订阅相匹配。服务器会将消息发送给订阅所匹配标签的每个客户端。需要注意的是， MQTT 中消息主题按照层级命名，使用 ‘/’ 进行分割。此外，主题中可以使用通配符进行多个主题或多层级的订阅，有两种常见的通配符：

        单层通配符 + ：单层通配符只能匹配一层的主题，例如： China/Beijing/+ ，可以匹配的只有 Beijing 这个主 题下面一层的主题，例如 Xicheng, DongCheng, Xuanwu 等等。

        多层通配符 # ：顾名思义，多层通配符就是可以匹配多个层级的主题，例如： China/# ，可以匹配到的主题可能有：China/Beijing/Dongcheng, China/Shanghai/PuDong ，等等。

        主题筛选器（Topic Filter）：一个对主题名通配符筛选器，在订阅表达式中使用，表示订阅所匹配到的多个主题。

        消息订阅：消息订阅者所具体接收的内容。

1.3、MQTT中的角色

        Publisher 和 Subscriber 为客户端，Broker 为服务器端，消息主题为消息类型，Broker 根据 Topic 过滤消息，并将消息向客户端推送。

MQTT 中用 QoS 表示服务质量， MQTT 协议中有三种服务质量 (QoS) ：

        1 ） QoS =0 ，至多一次，可能会出现丢包的情况，使用在对实时性要求不高的情况，例如，将此服务质量与通信环境传感器数据一起使用。 对于是否丢失个别读取或是否稍后立即发布新的读取并不重要。

        2 ） QoS =1, 至少一次，保证包会到达目的地，但是可能出现重包。

        3 ） QoS =2, 刚好一次，保证包会到达目的地，且不会出现重包的现象。

客户端：

        Publisher 和 Subscriber 都属于客户端。

        发布应用消息给其它相关的客户端。

        订阅以请求接受相关的应用消息。

        取消订阅以移除接受应用消息的请求。

        从服务端断开连接

服务器：

        服务器端即所谓的 MQTT Broker 服务器。

        接受来自客户端的网络连接。

        接受客户端发布的应用消息。

        处理客户端的订阅和取消订阅请求。

        转发应用消息给符合条件的已订阅客户端。

MQTT 提供的公共服务器端（ Broker ）有：

test.mosquitto.orgbroker.hivemq.comiot.eclipse.org

2、基于公共服务示例

        在次，选择使用EMQX提供的免费MQTT公共服务器，但同样可以选择其他任何MQTT broker。The Free Global Public MQTT Broker | Try Now | EMQ (emqx.com)

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

python库：pip install paho-mqtt=="1.6.1"

消息发布代码，pub.py

import time
import random
from paho.mqtt import client as mqtt_clientbroker = 'broker.emqx.io'
port = 1883
topic = "/flask/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_connectclient.connect(broker, port)return clientdef publish(client):msg_count = 0while 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 += 1def run():client = connect_mqtt()client.loop_start()publish(client)if __name__ == '__main__':run()

运行打印结果：

消息订阅代码，sub.py：

import random
from paho.mqtt import client as mqtt_clientbroker = 'broker.emqx.io'
port = 1883
topic = "/flask/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_connectclient.connect(broker, port)return clientdef 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_messagedef run():client = connect_mqtt()subscribe(client)client.loop_forever()if __name__ == '__main__':run()

运行打印结果：

        可以看到，发布和订阅成功。注意：每个客户端的ID唯一，不能重复！

3、基于Apollo服务示例

3.1、安装Apollo服务

        服务器端搭建：

Index of /dist/activemq/activemq-apollo/1.7.1

解压并打开目标如下：

        Apollo中间件其实是免安装的，我们只需要下载apache-apollo-1.7.1-windows-distro.zip,然后解压到某个文件夹就可以了。在这里我解压到D:\dist\apache-apollo-1.7.1。在apache-apollo-1.7.1/bin目录下打开终端执行命令：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版本即可。

        然后在生成的myapollo目录的bin目录下执行：pollo-broker.cmd run，结果如下：

        服务搭建成功，进入后台管理，打开网页，输入ip + : 61680 进入后台管理 ，默认用户名admin 密码 password，例如：http://127.0.0.1:61680

3.2、示例代码

本地服务基础信息：

host="127.0.0.1" 
port = 61613 
用户名：admin 
密码：password

python库：pip install paho-mqtt=="1.6.1"

发布主题，publish.py

import sys
import time
import paho.mqtt.client as mqttdef 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 += 1except KeyboardInterrupt:print("EXIT")client.disconnect()sys.exit(0)

订阅主题，subscribe.py

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

执行脚本后，正确大小消息。浏览器可见连接记录。

4、EMQX客户端工具

下载地址：MQTTX 下载选择适合您的平台，立即开始使用 MQTTX。icon-default.png?t=N7T8https://mqttx.app/zh/downloads。

4.1、公共服务测试消息接收

创建连接：

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

添加订阅：

主题为：/flask/mqtt

在MQTTX中发布消息：

测试成功。

4.2、自建服务测试消息接收

MQTT版本选择3.1.1，以下参数，与服务保持一致：

host="127.0.0.1" 
port = 61613 
用户名：admin 
密码：password

其它测试步骤，一样。（创建主题，测试）

5、EMQX代理服务器

        windows下搭建MQTT代理服务，与Apollo服务功能一样，更方便好用，下载地址：https://www.emqx.io/zh/downloads

        下载后解压，进入目录bin文件下，执行：emqx start 

        打开浏览器输入 http://localhost:18083 进入EMQ的web控制台，输入用户名：admin 密码：public 进行登录。登录进入后，界面如下：

至此，代理服务器已经创建完成！客户端就可以连接代理服务器进行消息的分发和订阅！

附录：

java1.8.1下载，地址：Java Downloads | Oracleicon-default.png?t=N7T8https://www.oracle.com/java/technologies/downloads/#java8-windows

参考：

1、嵌入式QT- QT使用MQTT

嵌入式QT- QT使用MQTT_qt mqtt-CSDN博客

2、Python实现通信协议（mqtt）5星，mqtt flask

【小沐学Python】Python实现通信协议（mqtt）_python mqtt-CSDN博客

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/679643.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

转转小程序数据处理

转转小程序数据处理

声明 本文章中所有内容仅供学习交流使用，不用于其他任何目的，抓包内容、敏感网址、数据接口等均已做脱敏处理，严禁用于商业用途和非法用途，否则由此产生的一切后果均与作者无关！wx a15018601872，x30184483x…
阅读更多...
【25届秋招备战C++】23种设计模式

【25届秋招备战C++】23种设计模式

【25届秋招备战C】23种设计模式 一、简介程序员的两种思维8大设计原则 二、具体23种设计模式2.1 创建型模式2.2 结构性模式2.3 行为型模式 三、常考模式的实现四、参考 一、简介 从面向对象谈起， 程序员的两种思维 底层思维:向下 封装：隐藏内部实现 多…
阅读更多...
FreeRTOS学习笔记-基于stm32（6）时间片调度实验

FreeRTOS学习笔记-基于stm32（6）时间片调度实验

1、什么是时间片调度 在任务优先级相同的时候，CPU会轮流使用相同的时间去执行它，即时间片调度。这个相同的时间就是时间片。而时间片的大小就是SysTick的中断周期（SysTick的中断周期可以修改）。 比如有三个相同优先级的任务在运行…
阅读更多...
【linux kernel】linux内核hid触摸源码hid-multitouch.c剖析

【linux kernel】linux内核hid触摸源码hid-multitouch.c剖析

文章目录 一、内核中通用hid触摸驱动二、probe过程剖析(1)hid_parse()函数(2)hid_hw_start()函数(3)hid_connect()函数三、hid-multitouch.c应用场景一、内核中通用hid触摸驱动 在linux内核中，为HID触摸面板实现了一个通用的驱动程序，位于/drivers/hid/hid-multitouch.c文件…
阅读更多...
在思科和华为上实现两个主机A,B A能ping通B,B不能ping通A

在思科和华为上实现两个主机A,B A能ping通B,B不能ping通A

1.华为实验的topo如下 常规状态下任意两台主机都是可以ping通的 此时的需求是PC4能ping通PC2和PC3但是PC2和PC3不能ping通PC4 这里需要用到ACL策略 在接口上调用 验证： PC4能ping通PC2和PC3 PC2和PC3不能ping通PC4 2.思科类似 正常情况下是都能互相ping通 加上ac…
阅读更多...
AI换脸原理(7)——人脸分割参考文献TernausNet: 源码解析

AI换脸原理(7)——人脸分割参考文献TernausNet: 源码解析

1、介绍 这篇论文相对来说比较简单，整体是通过使用预训练的权重来提高U-Net的性能，实现对UNet的改进。该方法也是DeepFaceLab官方使用的人脸分割方法。在介绍篇我们已经讲过了UNet的网络结构和设计，在进一步深入了解TernausNet之前，我们先简单回顾下UNet。 U-Net的主要结构…
阅读更多...
STC8增强型单片机开发——C51版本Keil环境搭建

STC8增强型单片机开发——C51版本Keil环境搭建

一、目标 了解C51版本Keil开发环境的概念和用途掌握C51版本Keil环境的安装和配置方法熟悉C51版本Keil开发环境的使用 二、准备工作 Windows 操作系统Keil C51 安装包（可以从Keil官网下载）一款8051单片机开发板 三、搭建流程 环境搭建的基本流程&#xf…
阅读更多...
吴恩达2022机器学习专项课程C2（高级学习算法）W1（神经网络）：2.5 更复杂的神经网络

吴恩达2022机器学习专项课程C2（高级学习算法）W1（神经网络）：2.5 更复杂的神经网络

目录 示例填写第三层的层数1.问题2.答案 公式：计算任意层的激活值激活函数 示例 层数有4层，不包括输入层。 填写第三层的层数 1.问题 你能把第二个神经元的上标和下标填写出来吗？ 2.答案 根据公式g(wxb)，这里的x对应的是上…
阅读更多...
【C语言】内存函数的概念，使用及模拟实现

【C语言】内存函数的概念，使用及模拟实现

Tiny Spark get dazzling some day. 目录 1. memcpy-- 函数原型-- 函数使用-- 函数的模拟实现 2.memmove-- 函数原型-- 函数使用-- 函数的模拟实现 3. memset-- 函数原型-- 函数使用-- 函数的模拟实现 4. memcmp-- 函数原型-- 函数使用-- 函数的模拟实现 1. memcpy 使用需包含…
阅读更多...
在国企分公司做信息宣传新闻投稿的经验分享

在国企分公司做信息宣传新闻投稿的经验分享

作为一名国企分公司的信息宣传工作者,我亲历了从传统投稿方式到数字化转型的全过程,这段经历既充满了挑战,也收获了成长。回首最初的日子,那些用邮箱投稿的时光,至今仍让我感慨万千。 初尝辛酸,邮箱投稿的艰难岁月 刚接手信息宣传工作时,我满腔热情,却很快被现实的冷水浇了个透…
阅读更多...
场景文本检测识别学习 day08（无监督的Loss Function、代理任务、特征金字塔）

场景文本检测识别学习 day08（无监督的Loss Function、代理任务、特征金字塔）

无监督的Loss Function（无监督的目标函数） 根据有无标签，可以将模型的学习方法分为：无监督、有监督两种。而自监督是无监督的一种无监督的目标函数可以分为以下几种： 生成式网络的做法，衡量模型的输出和固…
阅读更多...
【仪酷LabVIEW AI工具包案例】使用LabVIEW AI工具包+YOLOv5结合Dobot机械臂实现智能垃圾分类

【仪酷LabVIEW AI工具包案例】使用LabVIEW AI工具包+YOLOv5结合Dobot机械臂实现智能垃圾分类

‍‍🏡博客主页： virobotics(仪酷智能)：LabVIEW深度学习、人工智能博主 🎄所属专栏：『仪酷LabVIEW AI工具包案例』 📑上期文章：『【YOLOv9】实战二：手把手教你使用TensorRT实现YOLOv…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023