使用paho.mqtt.client实现MQTT Client连接EMQX Broker

概述

1 认识paho.mqtt.client

2 实现MQTT Client

2.1 功能介绍

2.2 paho.mqtt.client库函数介绍

2.3 MQTT Client实现

2.3.1 创建项目

2.3.2 编写MQTT Client代码

2.3.3 Log工具源码

2.4 功能测试代码实现

2.4.1 功能介绍

2.4.2 代码实现

3 测试

3.1 EMQX上创建Client

3.2 运行UserMqttClient

3.3 使用MQTT.fx订阅消息

概述

本文主要介绍使用paho.mqtt.client库实现一个MQTT Client，并使其连接到EMQX物联网平台。其中包括在EMQX创建项目的方法，配置参数的步骤。还是用该MQTT Client发布数据至EMQX，并使用MQTT.fx订阅Topic。MQTT Client也能订阅MQTT.fx发布的Topic数据。

1 认识paho.mqtt.client

paho.mqtt.client是一个Python MQTT客户端库，它提供了与MQTT代理进行通信的功能。MQTT是一种轻量级的消息传递协议，通常用于物联网应用中的设备间通信。

使用paho.mqtt.client，可以创建一个MQTT客户端，并使用其提供的方法来连接到MQTT代理，发布和订阅主题，接收和处理消息。

登录网站可以了解paho.mqtt.client的相关内容。登录地址如下：

https://mqtt.org/software/

2 实现MQTT Client

2.1 功能介绍

使用paho.mqtt.client编写一个 MQTT客户端，可以实现，消息订阅，发布功能。主要功能如下：

1）和Broker相关的参数通过配置文件来实现，不能写死在代码里

2）实现publish topic和subscriber topic功能

3）实时打印publish log和subscriber log

4）使用专用的logging库来处理log数据

5）使用class的方式来实现软件功能

2.2 paho.mqtt.client库函数介绍

在使用paho.mqtt.client库函数编写代码之前，必须对它所提供的重要接口函数的用法有一个清晰的认识，这对后面的如何使用这些接口非常重要。paho.mqtt.client提供了需要参数可供使用，这里只介绍一些，编程中必须用到的接口。其他函数的功能和用法可参考官方文档。

1）connect ()

连接MQTT Broker函数，函数原型如下：

参数介绍：

参数名称	功能介绍
host	MQTT Broker地址
port	连接Broker的端口号
keeplive	心跳包时间间隔

2）username_pw_set()

设置Client的用户名和对应的密码，函数原型如下：

参数介绍：

参数名称	功能介绍
username	Client的用户名
password	Client的用户名对应的密码

3）loop_start()

调用该接口，启动MQTT Client工作线程，这时就可以进行publish或者subscrib 消息，函数原型如下：

4）loop_stop()

调用该接口，销毁MQTT Client工作线程，函数原型如下：

5）subscribe()

订阅消息接口，函数原型如下：

参数介绍：

参数名称	功能介绍
topic	订阅的主题，例如：subscribe("my/topic", 2)
qos	消息的服务质量等级
options and properties	这两个参数在MQTT v5.0的版本中使用，当前版本不使用这两个参数

6）unsubscribe()

取消订阅消息接口，函数原型如下：

参数介绍：

参数名称	功能介绍
topic	取消订阅的主题，例如：unsubscribe("my/topic")
properties	这个参数在MQTT v5.0的版本中使用，当前版本不使用这个参数

7）publish()

发布消息接口，函数原型如下：

参数介绍：

参数名称	功能介绍
topic	发布消息的主题
payload	消息内容
qos	消息服务等级
retain	Broker是否保留消息
properties	这个参数在MQTT v5.0的版本中使用，当前版本不使用这个参数

2.3 MQTT Client实现

2.3.1 创建项目

本项目使用PyCharm 作为开发工具，运行代码前必须安装 paho.mqtt.client库。详细安装方法和步骤见官网文档。创建项目之后编写如下代码

2.3.2 编写MQTT Client代码

编写一个MQTTClient的user类，实现MQTT Client的基本功能，函数列表和介绍如下：

函数名	描述
start	注册回调函数和连接MQTT Client
stop	断开连接和销毁MQTT Client
usr_subscribe	订阅函数
usr_publish	发布消息函数
usr_unsubscribe	取消订阅函数
usr_on_message	订阅消息回调函数
usr_log_callback	log监控函数
receive_msg	接收消息函数

详细代码如下：

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @describe : mqtt handler
# @Time    : 2022/03/18 16:21
# @Author  : ming fei.tang
import logging
from queue import Queue

import paho.mqtt.client as mqtt

__all__ = ["MQTTClient"]


class MQTTClient:

    def __init__(self, host, port, qos, heartbeat, client_id, username, password):
        self.host = host
        self.port = port
        self.qos = qos
        self.queue = Queue()
        self.mqtt_client_id = client_id
        self.heartbeat = heartbeat
        self.username = username
        self.password = password
        self.mqtt_client = None

    def usr_on_message(self, user, data, msg):
        payload = msg.payload.decode('utf-8')
        payload = payload.replace('\n', '').replace('\r', '').replace(' ', '')
        logging.debug('subscribe: %s , payload: %s, QoS = %s' % (msg.topic, payload, msg.qos))

        self.queue.put(msg)

    def usr_subscribe(self, topic):
        self.mqtt_client.subscribe(topic, self.qos)
        logging.info('subscribe the topic: %s' % topic)

    def usr_unsubscribe(self, topic):
        self.mqtt_client.unsubscribe(topic)
        logging.info('unsubscribe %s' % topic)

    def receive_msg(self, timeout=None):
        logging.info('waiting for message.')
        if timeout is None:
            timeout = self.heartbeat
        return self.queue.get(timeout=timeout)

    def usr_publish(self, topic, payload, qos, retain=False):
        self.mqtt_client.publish(topic, payload, qos, retain)
        logging.debug('public topic = %s, payload = %s , qos = %s, retain = %s' % (topic, payload, qos, retain))

    def usr_log_callback(self, client, userdata, level, msg):
        # logging.info('public topic: %s ' % msg)
        pass

    def start(self):
        if self.mqtt_client is None:
            self.mqtt_client = mqtt.Client(client_id=self.mqtt_client_id)
            self.mqtt_client.on_log = self.usr_log_callback
            self.mqtt_client.on_message = self.usr_on_message
            self.mqtt_client.username_pw_set(self.username, self.password)
            self.mqtt_client.connect(self.host, self.port, self.heartbeat)
            self.mqtt_client.loop_start()
            logging.info("client('%s') is connected" % self.mqtt_client_id)
        else:
            logging.error("mqtt_client object is None")

    def stop(self):
        if self.mqtt_client is not None:
            self.mqtt_client.loop_stop()
            logging.info("client('%s')  is disconnected" % self.mqtt_client_id)
            self.mqtt_client.disconnect()
            self.mqtt_client = None

2.3.3 Log工具源码

创建logging_tool.py，编写如下代码：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# @Time    : 2020/4/9 13:05
# @Author  : ming fei.tang
# @File    : log tools
# ---------------------
import os
import logging
import datetime
import sys

import coloredlogs

__all__ = ["LogTool"]


class LogTool:
    def __init__(self):
        self.log_folder = 'clog' + os.sep + datetime.datetime.now().strftime('%Y%m%d')

    def setup_logging(self, level=logging.DEBUG, filename=None):
        if os.path.exists(self.log_folder) is False:
            os.makedirs(self.log_folder)

        log_filename = None
        if filename is not None:
            log_filename = self.log_folder + os.sep + filename

        log_format = '%(asctime)5s - %(levelname)5s - %(lineno)4s - %(filename)18s - %(message)s'
        if log_filename is not None:
            console = logging.StreamHandler(stream=sys.stdout)
            console.setLevel(logging.getLogger().level)
            console.setFormatter(logging.Formatter(log_format))
            logging.getLogger().addHandler(console)

        logging.basicConfig(filename=log_filename, level=level, format=log_format)
        coloredlogs.install(level=level, fmt=log_format, milliseconds=True)

    def remove_log_folder(self):
        if os.path.exists(self.log_folder) is False:
            os.remove(self.log_folder)

2.4 功能测试代码实现

2.4.1 功能介绍

测试函数主要完成功能：

1）连接EMQX服务器上的mqtt_user3 MQTT Client

2）发布消息至EMQX Broker

3) 订阅其他Client的消息，并通过log打印出来

2.4.2 代码实现

编写一个UserMqttClient的类，测试MQTTClient类中的方法，函数列表和介绍如下：

详细代码如下：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# @Time    : 2023/4/9 13:05
# @Author  : ming fei.tang
# @File    : mqtt client test
# ---------------------
import logging
import random
import time

from lib.MQTT_Client import MQTTClient
from lib.logging_tool import LogTool


class UserMqttClient(MQTTClient):
    def __init__(self, broker_address, mqtt_port, qos, heartbeat, client_ID, user_name, user_password):
        # prepare log file
        super().__init__(broker_address, mqtt_port, qos, heartbeat, client_ID, user_name, user_password)
        self.debug = LogTool()
        self.debug.setup_logging()
        self.debug.remove_log_folder()
        try:
            self.start()
        except Exception as e:
            logging.exception(e)
            assert False, e

    def load_para(self):
        pass

    def user_publish(self, topic, base_payload):
        while True:
            val = random.randint(1, 15) * 0.1
            val = base_payload + val
            val = '{:.2f}'.format(val)
            publish_payload = str(val)
            self.usr_publish(topic, publish_payload, 1, True)
            time.sleep(10)


if __name__ == '__main__':
    host = "192.168.1.11"
    port = 1883
    client_id = "paho.mqtt.client"
    username = "mqtt_user3"
    password = "123456"

    user_client = UserMqttClient(host, port, 0, 60, client_id, username, password)

    user_client.usr_subscribe("switch")
    user_client.usr_subscribe("attributes")

    user_client.user_publish(topic="temperature", base_payload=12)