Khi trí tuệ nhân tạo và dữ liệu lớn phát triển mạnh mẽ, Python đã trở thành một trong những ngôn ngữ lập trình phổ biến nhất nhờ vào sự gọn gàng trong cú pháp, thư viện phong phú và độ dốc học tập nhẹ nhàng. Điều này giúp Python dễ dàng hòa nhập vào các lĩnh vực như Internet vạn vật (IoT).

Có rất nhiều thư viện MQTT client tuyệt vời cho Python, nhờ đó chúng ta có thể xây dựng ứng dụng MQTT với lượng mã nguồn tối thiểu.

Bài viết này sẽ tổng hợp ba thư viện MQTT client phổ biến trong Python và so sánh chúng dựa trên các tiêu chí như sự phát triển, mức độ phức tạp khi sử dụng để hỗ trợ người đọc trong việc lựa chọn. Đồng thời, bài viết cũng cung cấp ví dụ đơn giản bằng Python mà bạn có thể sao chép và chạy trực tiếp (yêu cầu Python 3.5+ và cài đặt các gói phụ thuộc tương ứng).

paho-mqtt

paho-mqtt được xem là một trong những thư viện MQTT client mã nguồn mở hàng đầu cho Python. Được phát triển bởi Quỹ Eclipse, ngoài thư viện Python thì nó cũng hỗ trợ các ngôn ngữ lập trình chính như C++, Java, JavaScript, Golang,... Phiên bản Python hiện tại đã hỗ trợ MQTT 3.1 và 3.1.1, và phiên bản mới nhất đã hỗ trợ MQTT 5.0.

Dưới sự hỗ trợ của quỹ, nó được cập nhật mỗi năm một phiên bản. Phiên bản mới nhất tại thời điểm bài viết là 1.5.0 (phát hành tháng 8 năm 2019).

Trang chủ GitHub cung cấp hướng dẫn từ cơ bản đến chi tiết cho từng hàm, bao gồm cả những điều cần thiết cho người mới bắt đầu và người dùng nâng cao. Ngay cả khi gặp vấn đề vượt quá phạm vi, tìm kiếm trên Google cũng cho thấy gần 200.000 kết quả liên quan, đây là thư viện MQTT client phổ biến nhất hiện nay.

Được chú ý nhiều như vậy không chỉ vì mã nguồn ổn định mà còn do tính dễ sử dụng. Giao diện của Paho cực kỳ đơn giản và thanh lịch, chỉ cần một ít mã lệnh là có thể thực hiện việc đăng ký nhận tin và gửi tin qua MQTT.

Cài đặt

pip3 install paho-mqtt

hoặc

git clone https://github.com/eclipse/paho.mqtt.python
cd paho.mqtt.python
python3 setup.py install

Người nhận tin

import paho.mqtt.client as mqtt

# Hàm callback khi kết nối thành công
def on_connect(client, userdata, flags, rc):
    print(f"Kết nối thành công với mã {rc}")
    client.subscribe("$SYS/#")
    
# Hàm callback khi nhận được tin nhắn
def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))
    
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.emqx.io", 1883, 60)
client.loop_forever()

Người gửi tin

import paho.mqtt.client as mqtt
import time
def on_connect(client, userdata, flags, rc):
    print(f"Kết nối thành công với mã {rc}")
    
client = mqtt.Client()
client.on_connect = on_connect
client.connect("broker.emqx.io", 1883, 60)
for i in range(3):
    client.publish('a/b', payload=i, qos=0, retain=False)
    print(f"Gửi {i} tới chủ đề a/b")
    time.sleep(1)

client.loop_forever()

Thậm chí bạn có thể thực hiện cả việc đăng ký và gửi tin chỉ bằng một dòng lệnh:

import paho.mqtt.subscribe as subscribe

# Khi gọi hàm này, chương trình sẽ chờ đợi cho đến khi có tin nhắn gửi đến chủ đề paho/test/simple
msg = subscribe.simple("paho/test/simple", hostname="broker.emqx.io")
print(f"{msg.topic} {msg.payload}")

import paho.mqtt.publish as publish

# Gửi một tin nhắn
publish.single("a/b", "payload", hostname="broker.emqx.io")
# Hoặc gửi nhiều tin nhắn cùng lúc
msgs = [{'topic':"a/b", 'payload':"nhiều 1"}, ("a/b", "nhiều 2", 0, False)]
publish.multiple(msgs, hostname="broker.emqx.io")

HBMQTT

HBMQTT được xây dựng trên nền tảng Python asyncio, chỉ hỗ trợ giao thức MQTT 3.1.1. Do sử dụng thư viện asyncio nên người lập trình cần dùng phiên bản Python 3.4 trở lên.

Tốc độ xử lý của CPU nhanh hơn nhiều so với các thao tác I/O như đĩa hoặc mạng. Trong một luồng duy nhất, dù CPU xử lý nhanh đến đâu, khi gặp thao tác I/O thì vẫn phải dừng lại chờ hoàn tất, làm lãng phí thời gian.

Để giải quyết vấn đề này, Python đã giới thiệu tính năng I/O bất đồng bộ. Từ Python 3.4, asyncio chính thức được đưa vào thư viện chuẩn và từ Python 3.5, đã thêm từ khóa async/await. Người dùng có thể dễ dàng chuyển đổi hàm thành hàm bất đồng bộ bằng cách thêm từ khóa async trước tên hàm.

HBMQTT được xây dựng trên thư viện asyncio chuẩn. Nó cho phép người dùng thiết lập điểm ngắt rõ ràng, thông qua I/O bất đồng bộ, MQTT client có thể treo tác vụ hiện tại khi nhận hoặc gửi tin nhắn và tiếp tục xử lý tác vụ khác.

Tuy nhiên, HBMQTT ít được biết đến hơn. Tìm kiếm trên Google chỉ cho thấy khoảng 6.000 kết quả liên quan, trên Stack Overflow chỉ có 10 câu hỏi. Điều này có nghĩa là nếu chọn HBMQTT, bạn cần khả năng giải quyết vấn đề tốt.

Điều thú vị là HBMQTT cũng là một máy chủ MQTT. Bạn có thể khởi động bằng lệnh hbmqtt:

$ hbmqtt
[2020-08-28 09:35:56,608] :: INFO - Exited state new
[2020-08-28 09:35:56,608] :: INFO - Entered state starting
[2020-08-28 09:35:56,609] :: INFO - Listener 'default' bind to 0.0.0.0:1883 (max_connections=-1)

Cài đặt

pip3 install hbmqtt

hoặc

git clone https://github.com/beerfactory/hbmqtt
cd hbmqtt
python3 setup.py install

Người nhận tin

import logging
import asyncio
from hbmqtt.client import MQTTClient, ClientException
from hbmqtt.mqtt.constants import QOS_1, QOS_2

async def uptime_coro():
    C = MQTTClient()
    await C.connect('mqtt://broker.emqx.io/')
    await C.subscribe([
            ('$SYS/broker/uptime', QOS_1),
            ('$SYS/broker/load/#', QOS_2),
         ])
    try:
        for i in range(1, 100):
            message = await C.deliver_message()
            packet = message.publish_packet
            print(f"{i}:  {packet.variable_header.topic_name} => {packet.payload.data}")
        await C.unsubscribe(['$SYS/broker/uptime', '$SYS/broker/load/#'])
        await C.disconnect()
    except ClientException as ce:
        logging.error("Client exception: %s" % ce)
        
if __name__ == '__main__':
    formatter = "[%(asctime)s] %(name)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s"
    logging.basicConfig(level=logging.DEBUG, format=formatter)
    asyncio.get_event_loop().run_until_complete(uptime_coro())

Người gửi tin

import logging
import asyncio
import time
from hbmqtt.client import MQTTClient
from hbmqtt.mqtt.constants import QOS_0, QOS_1, QOS_2

async def test_coro():
    C = MQTTClient()
    await  C.connect('mqtt://broker.emqx.io/')
    tasks = [
        asyncio.ensure_future(C.publish('a/b', b'TEST MESSAGE WITH QOS_0', qos=QOS_0)),
        asyncio.ensure_future(C.publish('a/b', b'TEST MESSAGE WITH QOS_1', qos=QOS_1)),
        asyncio.ensure_future(C.publish('a/b', b'TEST MESSAGE WITH QOS_2', qos=QOS_2)),
    ]
    await asyncio.wait(tasks)
    logging.info("messages published")
    await C.disconnect()
    
if __name__ == '__main__':
    formatter = "[%(asctime)s] %(name)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s"
    logging.basicConfig(level=logging.DEBUG, format=formatter)
    asyncio.get_event_loop().run_until_complete(test_coro())

Xem thêm chi tiết tại tài liệu chính thức: https://hbmqtt.readthedocs.io/en/latest/.

gmqtt

gmqtt là một thư viện client mã nguồn mở do cá nhân phát triển. Mặc định hỗ trợ giao thức MQTT 5.0, nếu máy chủ MQTT không hỗ trợ 5.0 thì sẽ tự động hạ cấp xuống 3.1 và kết nối lại.

So với hai thư viện trước, gmqtt đang ở giai đoạn phát triển sơ khai, phiên bản hiện tại là 0.6.7. Tuy nhiên, đây là một trong những thư viện đầu tiên hỗ trợ MQTT 5.0, nên mức độ phổ biến trên mạng vẫn đáng kể.

Cũng được xây dựng trên thư viện asyncio, nên yêu cầu phiên bản Python 3.4 trở lên.

Cài đặt

pip3 install gmqtt

hoặc

git clone https://github.com/wialon/gmqtt
cd gmqtt
python3 setup.py install

Người nhận tin

import asyncio
import os
import signal
import time
from gmqtt import Client as MQTTClient

STOP = asyncio.Event()

def on_connect(client, flags, rc, properties):
    print('Đã kết nối')
    
def on_message(client, topic, payload, qos, properties):
    print(f'THU NHẬN TIN: {topic} {payload}')
    
def on_subscribe(client, mid, qos, properties):
    print('ĐÃ ĐĂNG KÝ')
    
def on_disconnect(client, packet, exc=None):
    print('Đã ngắt kết nối')
    
def ask_exit(*args):
    STOP.set()

async def main(broker_host):
    client = MQTTClient("id-client")
    
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_subscribe = on_subscribe
    client.on_disconnect = on_disconnect
    
    # Kết nối tới máy chủ MQTT
    await client.connect(broker_host)
    
    # Đăng ký chủ đề
    client.subscribe('TEST/#')
    
    # Gửi dữ liệu kiểm tra
    client.publish("TEST/A", 'AAA')
    client.publish("TEST/B", 'BBB')
    
    await STOP.wait()
    await client.disconnect()
    
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    
    loop.add_signal_handler(signal.SIGINT, ask_exit)
    loop.add_signal_handler(signal.SIGTERM, ask_exit)

    host = 'broker.emqx.io'
    loop.run_until_complete(main(host))

Người gửi tin

import asyncio
import os
import signal
import time
from gmqtt import Client as MQTTClient

STOP = asyncio.Event()

def on_connect(client, flags, rc, properties):
    print('Đã kết nối')
    client.subscribe('TEST/#', qos=0)
    
def on_message(client, topic, payload, qos, properties):
    print(f'THU NHẬN TIN: {topic}, {payload}')
    
def on_disconnect(client, packet, exc=None):
    print('Đã ngắt kết nối')
    
def ask_exit(*args):
    STOP.set()
    
async def main(broker_host):
    client = MQTTClient("id-client")
    
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    
    await client.connect(broker_host)
    
    client.publish('TEST/TIME', str(time.time()), qos=1)
    
    await STOP.wait()
    await client.disconnect()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    
    loop.add_signal_handler(signal.SIGINT, ask_exit)
    loop.add_signal_handler(signal.SIGTERM, ask_exit)
    
    host = 'broker.emqx.io'  
    loop.run_until_complete(main(host))

Cách chọn lựa

Sau khi giới thiệu ba thư viện MQTT client trong Python, hãy cùng xem xét cách chọn lựa phù hợp cho bản thân. Mỗi thư viện đều có ưu và nhược điểm riêng:

• paho-mqtt: Tài liệu tốt nhất, dễ hiểu, hỗ trợ mạnh từ quỹ. Tuy nhiên, tài liệu hiện tại chưa hỗ trợ MQTT 5.0.
• HBMQTT: Sử dụng thư viện asyncio để tối ưu độ trễ I/O mạng. Tuy nhiên, cú pháp khó đọc hơn, cũng không hỗ trợ MQTT 5.0.
• gmqtt: Cũng sử dụng asyncio, dễ đọc hơn HBMQTT, đặc biệt hỗ trợ MQTT 5.0. Nhưng phát triển chậm, tương lai chưa rõ ràng.

Vì vậy, khi lựa chọn, bạn có thể tham khảo các hướng dẫn sau:

• Nếu bạn đang phát triển bình thường và muốn áp dụng trong môi trường sản xuất, paho-mqtt là lựa chọn tối ưu nhất. Tính ổn định và dễ đọc mã code vượt trội so với hai thư viện còn lại. Khi gặp lỗi, tài liệu tốt và hàng ngàn kết quả tìm kiếm trên mạng cũng sẽ giúp bạn dễ dàng tìm ra giải pháp.
• Đối với những người đã thành thạo với thư viện asyncio, có thể thử HBMQTT và gmqtt.
• Nếu bạn muốn học tập, đóng góp cho dự án mã nguồn mở hoặc sử dụng MQTT 5.0, hãy thử gmqtt và đóng góp code cho dự án nhé.

Quyền tác giả: Bài viết này là tác phẩm gốc của EMQ, vui lòng ghi rõ nguồn khi tái sử dụng.

Liên kết gốc: https://www.emqx.io/cn/blog/comparision-of-python-mqtt-client