Python MQTT 协议使用规范

MQTT的基本使用规范

一、MQTT
MQTT(消息队列遥测传输)是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。该协议构建在TCP/IP协议之上。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

二、设计规范：

● （1）精简，不添加可有可无的功能；
● （2）发布/订阅（Pub/Sub）模式，方便消息在传感器之间传递；
● （3）允许用户动态创建主题，零运维成本；
● （4）把传输量降到最低以提高传输效率；
● （5）把低带宽、高延迟、不稳定的网络等因素考虑在内；
● （6）支持连续的会话控制；
● （7）理解客户端计算能力可能很低；
● （8）提供服务质量管理；
● （9）假设数据不可知，不强求传输数据的类型与格式，保持灵活性。

三、MQTT 特点:
MQTT协议是为大量计算能力有限，且工作在低带宽、不可靠的网络的远程传感器和控制设备通讯而设计的协议，它具有以下主要的几项特性：
1、使用发布/订阅消息模式，提供一对多的消息发布，解除应用程序耦合；
2、对负载内容屏蔽的消息传输；
3、使用 TCP/IP 提供网络连接；
4、有三种消息发布服务质量(qos)：
“至多一次”，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感 器数据，丢失一次读记录无所谓，因为不久后 还会有第二次发送。
“至少一次”，确保消息到达，但消息重复可能会发生。
“只有一次”，确保消息到达一次。这一级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果。
5、小型传输，开销很小（固定长度的头部是 2 字节），协议交换最小化，以降低网络流量；
6、使用 Last Will 和 Testament 特性通知有关各方客户端异常中断的机制。

mqtt.png

四、这里介绍一种mqtt服务器 -->eclipse-mosquitto
使用docker安装eclipse-mosquitto

1. 查询镜像docker search mosquitto 注意： 选择STARS最多的
2. 拉取镜像docker pull eclipse-mosquitto
3. 查看镜像docker images eclipse-mosquitto
4. 创建目录文件
   mkdir -p /mosquitto/config
   mkdir -p /mosquitto/data
   mkdir -p /mosquitto/log
5. 进入vi /mosquitto/config/mosquitto.conf 文件中
   写入以下内容
   persistence true
   persistence_location /mosquitto/data
   log_dest file /mosquitto/log/mosquitto.log
6. 为目录授权
   chmod -R 755 /mosquitto
   chmode -R 777 /mosquitto/log #日志目录要最大权限
7. 建立运行脚本mosquitto.sh
   docker run -it --name=mosquitto --privileged -p 1883:1883 -p 9001:9001 -v /mosquitto/config:/mosquitto/config/ -v /mosquitto/data/:/mosquitto/data -v /mosquitto/log/:/mosquitto/log -d eclipse-mosquitto
   启动命令这里注意一点就是-要区分中英文
8. 进入vim /mosquitto/config/mosquitto.conf文件，将以下命令填写进去
   关闭匿名模式
   allow_anonymous false
   指定密码文件
   password_file /mosquitto/config/pwfile.conf
9. 以上过程结束之后重启服务
   docker restart image_id

五、基于Python实现mqtt

创建一个客户端脚本
import json
import paho.mqtt.client as mqtt
import time

TASK_TOPIC = 'test'  # 客户端发布消息主题
client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
client = mqtt.Client(client_id, transport='tcp')

client.connect("服务器ip", 1883, 60)  # 此处端口默认为1883，通信端口期keepalive默认60
client.loop_start()

def clicent_main(message: str):
    """
    客户端发布消息
    :param message: 消息主体
    :return:
    """
    time_now = time.strftime('%Y-%m-%d %H-%M-%S', time.localtime(time.time()))
    payload = {"msg": "%s" % message, "data": "%s" % time_now}
    # publish(主题：Topic; 消息内容)
    # MQTT是基于pub/sub模式的，你可以subscribe这个clientID
    # ensure_ascii=False 解决中文乱码问题
    client.publish(TASK_TOPIC, json.dumps(payload, ensure_ascii=False))
    print("Successful send message!")
    return True

if __name__ == '__main__':
    msg = "我是一条测试数据！"
    clicent_main(msg)
     创建一个服务端脚本

import json
import sys
import time
import paho.mqtt.client as mqtt

REPORT_TOPIC = 'test'  # 主题

def on_connect(client, userdata, flags, rc):
    print('connected to mqtt with resurt code ', rc)
    client.subscribe(REPORT_TOPIC)  # 订阅主题


def on_message(client, userdata, msg):
    """
    接收客户端发送的消息
    :param client: 连接信息
    :param userdata:
    :param msg: 客户端返回的消息
    :return:
    """
    print("Start server!")
    payload = json.loads(msg.payload.decode('utf-8'))
    print(payload)

def server_conenet(client):
    client.on_connect = on_connect   # 启用订阅模式
    client.on_message = on_message   # 接收消息
    client.connect("服务器ip", 1883, 60)   # 链接
    # client.loop_start()  # 以start方式运行，需要启动一个守护线程，让服务端运行，否则会随主线程死亡
    client.loop_forever()  # 以forever方式阻塞运行。


def server_stop(client):
    client.loop_stop()     # 停止服务端
    sys.exit(0)

def server_main():
    client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
    client = mqtt.Client(client_id, transport='tcp')
    server_conenet(client)

if __name__ == '__main__':
    # 启动监听
    server_main()