devbox
知新_RL
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

【python】MQTT通信_/python/mqttls0tls1crudjtibq0

作者：知新_RL | 2024-03-06 18:28:04

赞

踩

/python/mqttls0tls1crudjtibq0

一、MQTT协议

MQTT协议几个概念解释一下，对MQTT协议熟悉的请跳过这部分

1、遗言消息

① 一旦连接到MQTT服务器，遗言消息就会被服务器托管，本客户端凡是非正常断开连接 服务器就会将本遗言发送给订阅该遗言消息的客户端，告知对方本客户端离线

② retain=False 新的client订阅是不会收到遗嘱消息的

2、服务质量（QoS）

  0:至多一次，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。

  1:至少一次，确保消息到达，但消息重复可能会发生。

  2:只有一次，确保消息到达一次。这一级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果。

3、连接返回代码

0: 连接成功

1: 连接失败 - 不正确的协议版本

2: 连接失败 - 无效的客户端标识符

3: 连接失败 - 服务器不可用

4: 连接失败 - 错误的用户名或密码

4、消息主题

① 大小写敏感

② # 符号是用于匹配主题中任意层级的通配符

5、Client ID可以自定义，但要注意不能重复

6、MQTT只能传输字符串数据

二、下载MQTT服务器端EMQX

1、EMQX下载链接: 免费下载、试用 EMQ 产品

2、选择开源版（当然如果条件允许也可以选择企业版），选择对应系统

 3、下载完成，解压gzip文件，进入解压目录，进入命令行，输入

./bin/emqx start

三、下载MQTT服务器端MQTTX（也可以不下载）

1、下载链接： MQTT 桌面客户端 - MQTT X | EMQ

2、使用MQTTX客户端：

① 获取IP地址：打开命令行，输入systeminfo

②打开MQTTX客户端，新建连接

名称不能与已建的连接重名

Client ID是自动填写的，可以不用管

服务器地址填写第一步获取到的IP地址

其他信息暂时不用填，点击右上角的连接，可以看到返回已连接 

 

四、Python代码

1、安装python MQTT第三方库

pip install paho.mqtt

2、简单的发布和订阅

        2.1 publish_basic.py

  1. # publish_basic.py
  2. import time
  3. import json
  4. import random
  5. from paho.mqtt import client as mqtt_client
  6.  
  7. broker = '192.168.***.***'  # mqtt代理服务器地址
  8. port = 1883
  9. keepalive = 60     # 与代理通信之间允许的最长时间段（以秒为单位）
  10. topic = "test/send"  # 消息主题
  11. client_id = f'python-mqtt-send-{random.randint(0, 1000)}'  # 客户端id不能重复
  12.  
  13. def connect_mqtt():
  14.     '''连接mqtt代理服务器'''
  15.     def on_connect(client, userdata, flags, rc):
  16.         '''连接回调函数'''
  17.         if rc == 0:
  18.             print("Connected to MQTT successfully!")
  19.  
  20.         else:
  21.             print("Failed to connect, return code {0}".format(rc))
  22.     # 连接mqtt代理服务器，并获取连接引用
  23.     client = mqtt_client.Client(client_id)
  24.     client.on_connect = on_connect
  25.     client.connect(broker, port, keepalive)
  26.     return client
  27.  
  28. def publish(client):
  29.     '''发布消息'''
  30.     while True:
  31.         '''每隔4秒发布一次信息'''
  32.         time.sleep(4)
  33.         # mqtt只能传输字符串数据
  34.         info = {
  35.             'msg1': 'Hello kristina',
  36.             'msg2': 'msg {0}'.format(random.randint(0, 1000))
  37.         }
  38.         msg = json.dumps(info)
  39.         # 默认retain=False，一个Topic只能有一个retained消息，后设置的会覆盖前面的消息
  40.         result = client.publish(topic=topic, payload=msg, qos=0, retain=True)
  41.         # 删除retained消息
  42.         # result = client.publish(topic=topic, payload=None, qos=0, retain=True)
  43.         if result[0] == 0:
  44.             print("Send {0} to topic {1}".format(msg, topic))
  45.         else:
  46.             print("Failed to send message {0} to topic {1}".format(msg, topic))
  47.  
  48. def run():
  49.     '''运行发布者'''
  50.     client = connect_mqtt()
  51.     # 运行一个线程来自动调用loop()处理网络事件, 非阻塞
  52.     client.loop_start()
  53.     publish(client)
  54.  
  55. if __name__ == '__main__':
  56.     run()

        2.2 subscribe_basic.py

  1. # subscribe_basic.py
  2. import random
  3. from paho.mqtt import client as mqtt_client
  4. import json
  5. from utils_logger import get_logger
  6.  
  7. broker = '192.168.***.***'  # mqtt代理服务器地址
  8. port = 1883
  9. keepalive = 60     # 与代理通信之间允许的最长时间段（以秒为单位）
  10. topic = "test/subcribe"  
  11. client_id = f'python-mqtt-subscribe-{random.randint(0, 1000)}'  # 可自定义，但要注意客户端id不能重复
  12.  
  13. def connect_mqtt():
  14.     '''连接mqtt代理服务器'''
  15.     def on_connect(client, userdata, flags, rc):
  16.         '''连接回调函数'''
  17.         # 响应状态码为0表示连接成功
  18.         if rc == 0:
  19.             print("Connected to MQTT successfully!")
  20.         else:
  21.             print("Failed to connect, return code {0}".format(rc))
  22.  
  23.     client = mqtt_client.Client(client_id)
  24.     client.on_connect = on_connect
  25.     client.connect(broker, port, keepalive)
  26.     return client
  27.  
  28. def subscribe(client: mqtt_client):
  29.     '''订阅主题并接收消息'''
  30.     def on_message(client, userdata, msg):
  31.         '''订阅消息回调函数'''
  32.         data = json.loads(msg.payload)  # data = 字典  #payload = json数据
  33.         print("Received message from topic {0} ".format(msg.topic))
  34.         print("The message have {0} information".format(len(data)))
  35.         print("The information is '{0}'".format(data))
  36.  
  37.     # 订阅指定消息主题
  38.     client.subscribe(topic=topic, qos=0)
  39.     # 将回调函数指派给客户端实例
  40.     client.on_message = on_message
  41.  
  42. def run():
  43.     # 运行订阅者
  44.     client = connect_mqtt()
  45.     subscribe(client)
  46.     #  运行一个线程来自动调用loop()处理网络事件, 阻塞模式
  47.     client.loop_forever()# 保持 loop()调用
  48.  
  49.  
  50. if __name__ == '__main__':
  51.     run()

3、进阶版发布和订阅（添加写入log，遗言主题以及上线下线消息）

        3.1 publish.py

  1. # publish.py
  2. import time
  3. import json
  4. import random
  5. from paho.mqtt import client as mqtt_client
  6. from utils_logger import get_logger
  7.  
  8.  
  9. broker = '192.168.***.***'  # mqtt代理服务器地址
  10. port = 1883
  11. keepalive = 60     # 与代理通信之间允许的最长时间段（以秒为单位）
  12. topic = "test/send"  # 消息主题
  13. will_topic = "test/will"  # 遗言消息主题
  14. client_id = f'python-mqtt-send-{random.randint(0, 1000)}'  # 客户端id不能重复
  15. # 遗言消息内容
  16. will_msg = {
  17.             "ID": f"{client_id}",
  18.             "stat": "Offline"
  19.         }
  20. # 上线消息内容
  21. online_msg = {
  22.             "ID": f"{client_id}",
  23.             "stat": "online"
  24.         }
  25. logfolder = 'logs'
  26. logfile = f'mqtt_publish.log'
  27. publish_logger = get_logger(logfolder, logfile)
  28.  
  29. def connect_mqtt():
  30.     '''连接mqtt代理服务器'''
  31.     def on_connect(client, userdata, flags, rc):
  32.         '''连接回调函数'''
  33.         if rc == 0:
  34.             # 上线后清除retained消息
  35.             result = client.publish(topic=topic, payload=json.dumps(online_msg), qos=0, retain=False)
  36.             publish_logger.info("Connected to MQTT successfully!")
  37.             if result[0] == 0:
  38.                 publish_logger.info("Send {0} to topic {1}".format(online_msg, topic))
  39.             else:
  40.                 publish_logger.error("Failed to send message {0} to topic {1}".format(online_msg, topic))
  41.  
  42.         else:
  43.             publish_logger.error("Failed to connect, return code {0}".format(rc))
  44.     # 连接mqtt代理服务器，并获取连接引用
  45.     client = mqtt_client.Client(client_id)
  46.     client.will_set(topic=will_topic, payload=json.dumps(will_msg), qos=0, retain=False)
  47.     client.on_connect = on_connect
  48.     client.connect(broker, port, keepalive)
  49.     return client
  50.  
  51. def publish(client):
  52.     '''发布消息'''
  53.     while True:
  54.         '''每隔4秒发布一次信息'''
  55.         time.sleep(4)
  56.         # mqtt只能传输字符串数据
  57.         info = {
  58.             'msg1': 'Hello kristina',
  59.             'msg2': 'msg {0}'.format(random.randint(0, 1000))
  60.         }
  61.         msg = json.dumps(info)
  62.         # 默认retain=False，一个Topic只能有一个retained消息，后设置的会覆盖前面的消息
  63.         result = client.publish(topic=topic, payload=msg, qos=0, retain=True)
  64.         # 删除retained消息
  65.         # result = client.publish(topic=topic, payload=None, qos=0, retain=True)
  66.         if result[0] == 0:
  67.             publish_logger.info("Send {0} to topic {1}".format(msg, topic))
  68.         else:
  69.             publish_logger.error("Failed to send message {0} to topic {1}".format(msg, topic))
  70.  
  71. def run():
  72.     '''运行发布者'''
  73.     client = connect_mqtt()
  74.     # 运行一个线程来自动调用loop()处理网络事件, 非阻塞
  75.     client.loop_start()
  76.     publish(client)
  77.  
  78. if __name__ == '__main__':
  79.     run()

        3.2 subscribe.py

  1. # subcribe.py
  2. import random
  3. from paho.mqtt import client as mqtt_client
  4. import json
  5. import datetime
  6. from utils_logger import get_logger
  7. # systeminfo
  8. broker = '192.168.***.***'  # mqtt代理服务器地址
  9. port = 1883
  10. keepalive = 60     # 与代理通信之间允许的最长时间段（以秒为单位）
  11. topic = "test/subcribe"  
  12.  
  13. logfolder = 'logs'
  14. logfile = f'mqtt_subscribe.log'
  15. subcribe_logger = get_logger(logfolder, logfile)
  16. client_id = f'python-mqtt-subscribe-{random.randint(0, 1000)}'  # 可自定义，但要注意客户端id不能重复
  17.  
  18. def connect_mqtt():
  19.     '''连接mqtt代理服务器'''
  20.     def on_connect(client, userdata, flags, rc):
  21.         '''连接回调函数'''
  22.         # 响应状态码为0表示连接成功
  23.         if rc == 0:
  24.             subcribe_logger.info("Connected to MQTT successfully!")
  25.         else:
  26.             subcribe_logger.error("Failed to connect, return code {0}".format(rc))
  27.  
  28.     client = mqtt_client.Client(client_id)
  29.     client.on_connect = on_connect
  30.     client.connect(broker, port, keepalive)
  31.     return client
  32.  
  33. def subscribe(client: mqtt_client):
  34.     '''订阅主题并接收消息'''
  35.     def on_message(client, userdata, msg):
  36.         '''订阅消息回调函数'''
  37.         data = json.loads(msg.payload)  # data = 字典  #payload = json数据
  38.         subcribe_logger.info("Received message from topic {0} ".format(msg.topic)) # msg.payload.decode()
  39.         subcribe_logger.info("The message have {0} information".format(len(data)))
  40.         subcribe_logger.info("The information is '{0}'".format(data))
  41.  
  42.     # 订阅指定消息主题
  43.     client.subscribe(topic=topic, qos=0)
  44.     # 将回调函数指派给客户端实例
  45.     client.on_message = on_message
  46.  
  47. def run():
  48.     # 运行订阅者
  49.     client = connect_mqtt()
  50.     subscribe(client)
  51.     #  运行一个线程来自动调用loop()处理网络事件, 阻塞模式
  52.     client.loop_forever() # 保持 loop()调用
  53.  
  54.  
  55. if __name__ == '__main__':
  56.     run()

        3.3 效果图

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/知新_RL/article/detail/200760
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号