Shulou(Shulou.com)06/02 Report--

How to understand the Python MQTT asynchronous framework HBMQTT, in view of this problem, this article introduces the corresponding analysis and solution in detail, hoping to help more partners who want to solve this problem to find a more simple and feasible method.

What is async

The speed of CPU is much faster than disk, network and other IO operations, and in a thread, no matter how fast CPU executes, when it encounters IO operations, it has to stop and wait for read and write to be completed, which undoubtedly wastes a lot of time.

To solve this problem, Python adds the feature of asynchronous IO. In Python 3. 4, asyncio was formally included in the standard library, and the async/await keyword was added to Python 3. 5. Users can easily add the async keyword in front of the function to make the function asynchronous.

Among Python's MQTT client libraries, HBMQTT is the earliest Python MQTT library to support asynchronous IO.

HBMQTT library

HBMQTT is an open source library based on Python, which implements the MQTT 3.1.1 protocol and has the following features:

Support for QoS 0, QoS 1, and QoS 2 messages

Client automatic reconnection

Support for TCP and WebSocket

Support for SSL

Support plug-in system

Next we will demonstrate how to use the Python MQTT asynchronous framework-HBMQTT to easily implement an asynchronous Demo with MQTT publish and subscribe capabilities.

Project initialization determines Python version

This project uses Python 3.6for development testing. Readers can confirm the version of Python with the following command.

Because you need to use the async keyword, you need to make sure that the Python version is no less than Python 3.5

➜~ python3-- versionPython 3.6.7 installs the HBMQTT library using Pip

Pip is the package management tool of Python, which provides the functions of finding, downloading, installing, and uninstalling Python packages.

Pip3 install-I https://pypi.doubanio.com/simple hbmqtt connects to the MQTT server

This article will use the free public MQTT server provided by EMQ X, which is based on EMQ X's MQTT Internet of things cloud platform. The server access information is as follows:

Broker: broker.emqx.io

TCP Port: 1883

Websocket Port: 8083

First, import the MQTT client library.

From hbmqtt.client import MQTTClientclient = MQTTClient () # Connect server client.connect ('mqtt://broker.emqx.io/') # disconnect client.disconnect ()

Write asynchronously as follows:

Async def test_pub (): client = MQTTClient () await client.connect ('mqtt://broker.emqx.io/') await client.disconnect () publish the message

The publish message function is the publish function of the MQTTClient class.

The three parameters of the client = MQTTClient () # function are the subject, the message content, and QoSclient.publish ('a _

Write asynchronously as follows:

Async def test_pub (): client = MQTTClient () await Client.connect ('mqtt://broker.emqx.io/') await asyncio.gather (client.publish (' a b'TEST MESSAGE WITH QOS_0', qos=QOS_0), client.publish ('a b'TEST MESSAGE WITH QOS_1', qos=QOS_1), client.publish ('a b'TEST MESSAGE WITH QOS_1', qos=QOS_1), b'TEST MESSAGE WITH QOS_2' Qos=QOS_2)) logging.info ("messages published") await Client.disconnect ()

In this code, we put three message-sending functions into the task list of asyncio, and they will be run in turn. When all tasks are complete, disconnect.

Subscribe to messages

The subscription message function is the subscribe function in the MQTTClient class.

Client = MQTTClient () # subscribe to client.subscribe ([('topic/0', QOS_0), (' topic/1', QOS_1),]) # Unsubscribe to client.unsubscribe ([('topic/0', QOS_0),]

Write asynchronously as follows:

Async def test_sub (): client = MQTTClient () await client.connect ('mqtt://broker.emqx.io/') await client.subscribe ([(' a QOS_1),]) for i in range (0 10): message = await client.deliver_message () packet = message.publish_packet print (f "{I}: {packet.variable_header.topic_name} = > {packet.payload.data}") await client.disconnect ()

In this code, we set await wait when we receive the message. When the code executes to the following location, CPU will first perform other tasks until a message is delivered, and then print it.

Message = await client.deliver_message ()

Eventually, the program waits for 10 messages to be received and then closes the connection.

Full code message subscription code # sub.py# python 3.6+import asyncioimport loggingfrom hbmqtt.client import MQTTClientfrom hbmqtt.mqtt.constants import QOS_1async def test_sub (): client = MQTTClient () await client.connect ('mqtt://broker.emqx.io/') await client.subscribe ([((' a) QOS_1),]) for i in range (0 10): message = await client.deliver_message () packet = message.publish_packet print (f "{I}: {packet.variable_header.topic_name} = > {packet.payload.data}") await client.disconnect () if _ name__ = ='_ _ main__': formatter = "[% (asctime) s]% (name) s {% (filename) SV% (lineno) d}% (levelname) S -% (message) s "logging.basicConfig (level=logging.INFO) Format=formatter) asyncio.run (test_sub ()) message release code # pub.py# python 3.6+import asyncioimport loggingfrom hbmqtt.client import MQTTClientfrom hbmqtt.mqtt.constants import QOS_0, QOS_1, QOS_2async def test_pub (): client = MQTTClient () await client.connect ('mqtt://broker.emqx.io/') await asyncio.gather (client.publish (' a Universe, b'TEST MESSAGE WITH QOS_0', qos=QOS_0) Client.publish ('a _ Qos=QOS_2) logging.info ("messages published") await client.disconnect () if _ _ name__ = ='_ _ main__': formatter = "[% (asctime) s]% (name) s {% (filename) SSV% (lineno) d}% (levelname) s -% (message) s" logging.basicConfig (level=logging.INFO, format=formatter) asyncio.run (test_pub ()) Test message release

Running the MQTT message publishing code, we will see that the client connects successfully and publishes the message successfully.

The following is that the MQTT X client successfully received the message issued by the HBMQTT client:

Message subscription

Run the MQTT message subscription code, and we will see that the client connects successfully and the client is waiting for the message to enter

Connect to broker.emqx.io using the MQTT X client, and then send a message 10 times to the subject a _ b

When we get back to the terminal, we see that the client receives and prints the message, and actively exits the program after receiving 10 messages.

At this point, we have completed the connection of the HBMQTT library to the public MQTT server, and implemented the connection, message publishing and subscription between the test client and the MQTT server. By using Python asynchronous IO to send and receive messages, we can help us achieve a more efficient MQTT client.

This is the answer to the question on how to understand the Python MQTT asynchronous framework HBMQTT. I hope the above content can be of some help to you. If you still have a lot of doubts to be solved, you can follow the industry information channel for more related knowledge.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.