MQTT for python 使用手册

╰半夏微凉° 2022-12-03 08:29 114阅读 0赞

# MQTT #

是一种基于发布/订阅模式的 **SNMP(轻量级互联网网消息传输协议)**  
可以用很少的代码和带宽为互联网设备提供可靠的消息服务  
此处 — [官方文档][Link 1]

# MQTT for Python #

> 在python项目中使用mqtt服务

## 方法 ##

1.  **connect( host, port=1883, keepalive=60 bind\_address="" )**  
    — 以阻塞的方式进行连接
    
    host：代理主机名 or IP  
    port：端口  
    keepalive：代理通信的最长时间. 如果没有交换其他消息, 则控制客户端向代理服务器发送 ping 的速率  
    bing\_address：多个接口绑定本地 ip 地址
2.  **connect\_async( host, port=1883, keepalive=60, bind\_address="" )**  
    — 以异步非阻塞的方式进行连接
3.  **reconnect()**  
    — 使用之前的配置信息从新连接, 但是必须在 connect\*() 之前调用
4.  **disconnect()**  
    — 断开连接, 将不会等待所有排队的消息被发送，以确保所有邮件都已发送
5.  **loop( timeout=1.0, max\_packets=1 )**  
    — 定期调用网络处理
6.  **loop\_start() / loop\_stop()**  
    — 实现网络循环的线程接口, 可以控制线程的启动和结束
    
        mqttc.connect("mqtt.eclipse.org")
        mqttc.loop_start()
        
        while True:
           temperature = sensor.blocking_read()
           mqttc.publish("paho/temperature", temperature)
7.  **loop\_forever()**  
    — 网络循环的阻塞形式, 在调用 disconnect() 之前不会停止
8.  **publish( topic, payload=None )**  
    — 向客户端代理发送消息
    
    topic：要发送的主题  
    payload：发送的信息
9.  **subscribe( topic, qos=0 )**  
    — 订阅 topic, 如果订阅1个, qos可以忽略
    
    topic：想要订阅的主题  
    [qos 是什么][qos]
    
        # 例子
        subscribe("my/topic")
        subscribe(("my/topic", 1))
        subscribe([("my/topic", 0), ("another/topic", 2)])
10. **unsubscribe( topic )**  
    — 取消订阅
11. **username\_pw\_set( username, password=None )**  
    — 在 `connect()`之前设置`client`的用户名和密码，依据MQTT配置的`mqtt_acl`与`mqtt_user`表中的ACL规则与用户信息进行用户验证。  
    只要MQTT开启了ACL验证， 就必须登录验证。
12. **on\_connect( client, userdata, flags, rc )**  
    — 当代理响应我们连接请求时调用
    
    client：客户端对象  
    userdata：在 Client()\\user\_data\_set() 中设置的私有用户数据  
    flags：代理服务器的响应标志  
    rc：连接结果
    
    > flags: 本身是一个 dict 字典, `flags['session present']`,  
    > 如果具clean session = 0, 客户端重新连接到它先前已连接的代理，则此标志指示代理是否仍然具有该客户端的会话信息。 如果为1，则会话仍然存在。
    
    > rc : 连接状况  
    > – 0：连接成功  
    > – 1：连接拒绝-不正确的协议版本  
    > – 2：连接拒绝-无效客户标识符  
    > – 3：连接拒绝-服务器不可用  
    > – 4：连接拒绝-错误用户名或密码  
    > – 5：连接拒绝-未授权  
    > – 6-255：当前未使用。
    
        def on_connect(client, userdata, flags, rc):
            print("Connection returned result: "+connack_string(rc))
        
            mqttc.on_connect = on_connect
            ...
13. **on\_disconnect( client, userdata, rc)**  
    — 当客户端断开连接时使用
    
    rc：断开状态, 如果是 0 , 则是调用 `disconnect()`断开的, 如果是其他任何值, 则表示意外断开
    
        def on_disconnect(client, userdata, rc):
            if rc != 0:
                print("Unexpected disconnection.")
        
        mqttc.on_disconnect = on_disconnect
        ...
14. **on\_message( client, userdata, message)**  
    — 当在订阅的主题上收到了消息, 并且与现在的主题筛选器回调不匹配时使用
    
    message 是一个 `MQTTMessage`实例, 它的属性有`topic`, `payload`,`qos`,`retain`
    
        def on_message(client, userdata, message):
           print("Received message '" + str(message.payload) + "' on topic '"
               + message.topic + "' with QoS " + str(message.qos))
        
        mqttc.on_message = on_message
15. **on\_publish(client, userdata, mid)**  
    — 发布的回调函数
16. **on\_subscribe(client, userdata, mid, granted\_qos)**  
    — 订阅的回调函数
17. **on\_unsubscribe(client, userdata, mid)**  
    — 取消订阅的回调函数
18. **on\_log(client, userdata, level, buf)**  
    — 客户端有日志信息时调用, 能与 Python Log 同时使用
    
    等级包括：`MQTT_LOG_INFO`, `MQTT_LOG_NOTICE`, `MQTT_LOG_WARNING`, `MQTT_LOG_ERR`, `MQTT_LOG_DEBUG`

## 安装 ##

> pip install paho-mqtt

## 使用 ##

接下来的例子通过 EMQ X 提供的免费 MQTT服务器 来实施

参数信息:

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

#### 连接 MQTT 服务器 ####

1.  导入 Paho MQTT 客户端
    
        from paho.mqtt import client as mqtt_client
2.  设置 MQTT broker 连接参数
    
    > 设置 Broker 的链接地址, 端口, topic, 生成随机client\_ID
    
        broker = 'broker.emqx.io'
        port = 1883
        topic = "/python/mqtt"
        client_id = f'python-mqtt-{random.randint(0, 1000)}'
3.  写入连接函数
    
    > 编写连接回调函数 **on\_connect**, 该函数将在 Client 连接后被调用, 可以根据函数中的参数 **rc** 来判断是否连接成功
    
        def connect_mqtt():
            # 编写回调函数
            def on_connect(client, userdata, flags, rc):
                if rc == 0:
                    print('连接成功!')
                else:
                    print('连接失败!, rc为%d', rc)
            # 设置 Client 
            client = mqtt_client.Client(client_id)
            client.on_connect = on_connect
            client.connect(broker, port)
            return client

#### 发布消息 ####

> 借助 Client.publish(topic, msg), 可以将 msg 发布到所有订阅了 topic 的客户端中

def pubush(client):
        msg_count = 0
        while True:
            time.sleep(1)
            msg = f"Publish msg :{msg_count}"
            result = client.publish(topic, msg)
            # 返回值 result: [0, 1]
            status = result[0]
            if status == 0:
                print(f"给 topic '{topic}' 发送了消息: '{msg}' ")
            else:
                print(f'给 topic {topic} 发送消息失败!')
            count += 1

#### 订阅消息 ####

> 通过 Client.subscribe(topic) , 会将订阅的 topic 下的所有消息接收

def subscribe(client):
        def on_message(client, userdata, msg):
            print(f"从 '{msg.topic}' 接收到的消息是 '{msg.payload.decode()}'")
        client.subscribe(topic)
        client.on_message = on_message

#### 完整代码 ####

import random
    import time
    
    from paho.mqtt import client as mqtt_client
    
    broker = 'broker.emqx.io'
    port = 1883
    topic = "/python/mqtt"
    
    
    class MQTT:
        def __init__(self, mqtt_type):
            self.client_id = f'python-mqtt-{random.randint(0, 1000)}'
            self.client = mqtt_client.Client(client_id=self.client_id)
            self.client.on_connect = self.on_connect
            self.client.connect(broker, port)
    
            self.mqtt_type = mqtt_type
            if self.mqtt_type == 'pub':
                self.client.on_publish = self.on_publish
            else:
                self.client.on_message = self.on_message
    
        def on_connect(self, client, userdata, flags, rc):
            # 如果是订阅, 则在连接时进行订阅
            if self.mqtt_type == 'subs':
                self.client.subscribe(topic=topic)
    
    
        def on_message(self, client, userdata, msg):
            print(f"从 '{msg.topic}' 接收到的消息是 '{msg.payload.decode()}'")
    
        def on_publish(self):
            msg_count = 0
            while True:
                time.sleep(1)
                msg = f"Publish msg :{msg_count}"
                result = self.client.publish(topic, msg)
                # 返回值 result: [0, 1]
                status = result[0]
                if status == 0:
                    print(f"给 topic '{topic}' 发送了消息: '{msg}' ")
                else:
                    print(f'给 topic {topic} 发送消息失败!')
                msg_count += 1
    
        # 启动网络循环
        def forever(self):
            self.client.loop_forever()

#### 开启订阅和发布 ####

from mqtt import MQTT
    from threading import Thread
    
    
    def run(mqtt_type):
        if mqtt_type == 'subs':
            mqtt = MQTT(mqtt_type = 'subs')
            mqtt.forever()
        else:
            mqtt.MQTT(mqtt_type = 'pub')
            mqtt.on_publish()
            
    def main():
        subs = Thread(target = run, args = ('subs',))
        pub = Thread(target = run, args = ('pub',))
        subs.start()
        pub.start()
        subs.join()
        pub.join()
    
    if __name__ == '__main__':
        main()

#### 结果 ####

> 给 topic '/python/mqtt' 发送了消息: 'Publish msg :0'  
    > 从 '/python/mqtt' 接收到的消息是 'Publish msg :0'
    > 给 topic '/python/mqtt' 发送了消息: 'Publish msg :1'  
    > 从 '/python/mqtt' 接收到的消息是 'Publish msg :1'
    > 给 topic '/python/mqtt' 发送了消息: 'Publish msg :2'  
    > 从 '/python/mqtt' 接收到的消息是 'Publish msg :2'
    > 给 topic '/python/mqtt' 发送了消息: 'Publish msg :3'  
    > 从 '/python/mqtt' 接收到的消息是 'Publish msg :3'

## 其他 ##

#### 全局函数 ####

> client 模块提供了一些全局函数来辅助使用

*  **topic\_matches\_sub( sub, topic )**  
    — 匹配 topic(主题) 是否匹配 subscription(订阅), 也就是`topic in subscription`
    
    topic : `foo/bar` 匹配 : `foo/#`或者`+/bar`
 *  **error\_string(mqtt\_errno)**  
    — 返回与PAHO MQTT错误号关联的错误字符串。

#### Publish ####

> 允许一次简单的发送, 发送信息后就关闭连接, 包含`single`和`multiple`方法

*  **Single**

single(topic, payload=None, qos=0, retain=False,hostname="localhost",
           port=1883, client_id="", keepalive=60, will=None, auth=None, tls=None,
           protocol=mqtt.MQTTv311, transport="tcp")

— will : dict\{\}, 存放的是发送数据, `topic`是必须的参数

will = { 'topic':<topic>, 'payload':<payload>, 'qos':<qos>,  'retain':<retain>}

— auth : dict\{\}, 包含身份验证的信息, `username`是必须的

Auth={ 'username': <username>, 'Password': <Password>}

— tls : dict\{\}, 客户端的 TLS 参数, `ca_certs`是必须的参数

tls = { 'ca_certs': <ca_certs>, 'certfile': <certfile>, 'keyfile': <keyfile> , 'tls_version': <tls_version>, 'ciphers': <ciphers>}

— protocol: 选择使用MQTT的协议版本, `MQTTv31` 或者 `MQTTv311`

**一个 Single 发布的例子**

import paho.mqtt.publish as publish
    
    publish.single("paho/test/single", "payload", hostname="mqtt.eclipse.org")

*  **Multiple**  
    — 向代理发送多条数据, 之后断开连接

multiple(msgs, hostname="localhost", port=1883, client_id="", keepalive=60,
             will=None, auth=None, tls=None, protocol=mqtt.MQTTv311, transport="tcp")

msgs 有两种形式

*  dict, 如果是字典, 只有 `topic` 是必须的参数

msg = { 'topic': <topic>, 'payload': <payload>, 'qos': <qos>, 'retain':<retain>};

*  tuple, 如果是元组, 所有参数都需要填写, `qos`默认为0, `retain`默认为 False

msg = (<topic>, <payload>, <qos>, <retain>);

**一个 Mutiple 发布的例子**

import paho.mqtt.publish as publish
    
    msgs = [{ 'topic':"paho/test/multiple", 'payload':"multiple 1"},
             ("paho/test/multiple", "multiple 2", 0, False)]
    publish.multiple(msgs, hostname="mqtt.eclipse.org")

#### Subscribe ####

> 这个模块提供了一些函数能够直接的订阅和处理消息, 包含`simple`和`callback`方法

*  **Simple**

simple(topics, qos=0, msg_count=1, retained=False, hostname="localhost",
            port=1883, client_id="", keepalive=60, will=None, auth=None, tls=None,
            protocol=mqtt.MQTTv311)

**一个 Simple 订阅的例子**

import paho.mqtt.subscribe as subscribe
    
    msg = subscribe.simple("paho/test/simple", hostname="mqtt.eclipse.org")
    print("%s %s" % (msg.topic, msg.payload))

*  **Callback**

> 订阅一组主题，并使用用户提供的回调函数处理接收的消息。

callback(callback, topics, qos=0, userdata=None, hostname="localhost",
             port=1883, client_id="", keepalive=60, will=None, auth=None, tls=None,
             protocol=mqtt.MQTTv311)

**一个 Callback 订阅的例子**

import paho.mqtt.subscribe as subscribe
    
    def on_message_print(client, userdata, message):
        print("%s %s" % (message.topic, message.payload))
    
    subscribe.callback(on_message_print, "paho/test/callback", hostname="mqtt.eclipse.org")

[Link 1]: https://pypi.org/project/paho-mqtt/
[qos]: https://www.jianshu.com/p/8b0291e8ee02