MQTT消息框架paho-mqtt与emqx安装部署与启动，python

Posted 2022-10-25 zhangphil

MQTT消息框架paho-mqtt与emqx安装部署与启动，python

(1)安装MQTT消息框架paho-mqtt

pip install paho-mqtt

(2)写一个mqtt的消息订阅端sub.py ：

import json

import paho.mqtt.client as mqtt

"""rc 值	
0	连接成功
1	协议版本错误
2	无效的客户端标识
3	服务器无法使用
4	错误的用户名或密码
5	未经授权
"""
def on_connect(client, userdata, flags, rc):
    return rc


def on_message(client, userdata, msg):
    data = msg.payload
    print(json.loads(data))


def main():
    client = mqtt.Client()
    client.username_pw_set(username='', password='')
    client.on_connect = on_connect
    client.on_message = on_message
    rc = client.connect(host='127.0.0.1', port=9999, keepalive=60)
    print('rc', rc)
    if rc == 0:
        client.subscribe('my_topic')
    else:
        print('连接失败')
        raise Exception

    # 无限循环收发消息
    client.loop_forever()


if __name__ == '__main__':
    main()

（2）写一个mqtt的消息发布端pub.py：

import json
import random
import time

import paho.mqtt.client as mqtt


def publish():
    client = mqtt.Client()
    client.username_pw_set('', '')
    client.connect(host='127.0.0.1', port=9999)

    for i in range(5):
        msg = 
            'count': i,
            'type': 1,
            'msg': random.randint(0, 100),
            'time': time.strftime('%Y-%m-%d %H:%M:%S')
        
        client.publish(topic='my_topic', payload=json.dumps(msg), retain=False)


if __name__ == '__main__':
    publish()

（3）在mqtt的架构中，需要部署一个消息的中继broker，这里选择EMQX。下载emqx的服务器端，安装部署emqx服务器，下载Windows服务器端：

Download EMQXRun anywhere in private, hybrid, and public clouds. Download the latest EMQX and efficiently connect tens of millions of IoT devices via an EMQX cluster.https://www.emqx.io/downloads

下载解压，找到bin目录，在当前目录直接右键启动Windows的powershell，输入命令启动emqx服务器端：

./emqx start

如果是关闭，命令是：

./emqx stop

（4）到这里emqx作为mqtt的服务器端启动了，emqx默认的主页是：http://localhost:18083/

打开后，需要输入用户名和秘密，初始的用户名是 admin，密码 public

进去后是这样的web端页面

在emqx里面配置一个listener:

这里主要配置两个：tcp和绑定的端口，本例是9999。

（5）开始发布/订阅消息。

首先启动sub.py，然后在运行pub.py，在sub.py的控制台可以看到，收到了订阅的topic为“my_topic的消息”：

在web的dashboard里面，可以看到my_topic已经录入：

（6）关于发布消息，如果只是简单的发送/发布一条消息，则可以使用简化的方式：

import paho.mqtt.publish as sender


def pub():
    sender.single(
        topic='my_topic',
        payload=json.dumps('msg': 'ok'),
        hostname='127.0.0.1',
        port=9999,
        client_id='ip address:127.0.0.1',
        transport='tcp')

（7）在标准的publish的函数里面，有一个参数为

publish(self, topic, payload=None, qos=0, retain=False, properties=None)

retain值如果设置为True，则mqtt服务器端将保持发布者发布的最后一条消息，每当订阅者启动、连接到mqtt服务器端（emqx）时候，mqtt服务器端就会发给连接者最后的那条消息：“last known good”原则。默认False，相当于服务器端不记忆发布者发布的消息。