文章目錄
- 玩轉物聯網只需十行代碼，可它為何悄悄停止維護
- 1 背景：MQTT 遇上 asyncio，為什么選 hbmqtt？
- 2 hbmqtt 是什么？
- 3 安裝：一行命令，但別裝最新
- 4 五大核心 API：10 行代碼跑通發布訂閱
- 5 實戰五連擊：代碼復制即可跑
- 5.1 萬級消息吞吐——異步批量發布
- 5.2 斷網自恢復——監聽重連事件
- 5.3 本地 Broker——單元測試不依賴外網
- 5.4 WebSocket 直播——網頁秒級訂閱
- 5.5 SSL 上云——一參開啟加密
- 6 三大常見坑：報錯 + 原因 + 急救方案
- 7 總結：遺產代碼安全駕駛指南
玩轉物聯網只需十行代碼，可它為何悄悄停止維護
本文帶你速通 Python 老牌異步 MQTT 庫 hbmqtt——從安裝到踩坑，從入門到放棄（以及平滑遷移）。讀完你將明白：
它為何曾是“高并發 IoT 第一神器”，又為何在 2023 年正式謝幕；如何在遺留項目里繼續安全駕駛；新項目又該投奔誰。
1 背景：MQTT 遇上 asyncio，為什么選 hbmqtt？
- 物聯網場景動輒萬級長連接、低帶寬、高延遲，MQTT 憑借“輕量級發布/訂閱”成為事實標準。
- Python 側早期全是同步阻塞客戶端（Paho），一條網絡抖動就卡死整個線程。
- hbmqtt 率先100% 協程化，單進程輕松撐住10k 并發；還自帶Broker 實現，本地測試一條龍。
- 但！官方已宣布停止維護，社區分叉出繼任者 amqtt。
? 老項目仍可用，新項目建議直接上 amqtt 或 gmqtt。
2 hbmqtt 是什么？
一句話：純 Python 寫的異步 MQTT 3.1.1 客戶端 + 代理雙角色庫，基于 asyncio，支持 QoS0-2、WS/WSS、自動重連、插件鑒權。
3 安裝：一行命令，但別裝最新
# 官方最終穩定版
python -m pip install "hbmqtt==0.9.6"
?? 不指定版本會拉到 0.10+ 預發行版，API 已變且含未修復 Bug。
4 五大核心 API：10 行代碼跑通發布訂閱
import asyncio
from hbmqtt.client import MQTTClientasync def demo():client = MQTTClient() # 1. 創建客戶端await client.connect('mqtt://broker.emqx.io:1883') # 2. 連接await client.subscribe([('demo/hi', 1)]) # 3. 訂閱await client.publish('demo/hi', b'hello', qos=1) # 4. 發布msg = await client.deliver_message() # 5. 收消息print('Got:', msg.publish_packet.payload.data)await client.disconnect() # 6. 優雅斷開asyncio.run(demo())
| 行 | 作用 |
|---|---|
| 1 | 事件循環入口 |
| 3 | 生成異步客戶端實例 |
| 4 | TCP 連接公網免費 broker |
| 5 | 訂閱列表，支持通配符 +/ # |
| 6 | 發布字節串，必須 bytes |
| 7 | 阻塞等待下一條消息（協程不會卡線程） |
| 9 | 釋放 TCP / 會話資源 |
5 實戰五連擊：代碼復制即可跑
5.1 萬級消息吞吐——異步批量發布
async def bulk_publish(n=10_000):cli = MQTTClient()await cli.connect('mqtt://127.0.0.1:1883')# 使用 asyncio.gather 把發布變成“并發煙花”await asyncio.gather(*(cli.publish('load/test', f'pkt{i}'.encode(), qos=0) for i in range(n)))await cli.disconnect()
5.2 斷網自恢復——監聽重連事件
client = MQTTClient(config={'auto_reconnect': True, 'reconnect_max_interval': 5})
# 網絡抖動后自動重連，QoS1/2 消息不丟
5.3 本地 Broker——單元測試不依賴外網
from hbmqtt.broker import Broker
import yaml, asyncioasync def start_local():config = {'listeners': {'default': {'type': 'tcp', 'bind': '0.0.0.0:1883'}}}broker = Broker(yaml.safe_dump(config))await broker.start()await asyncio.sleep(60) # 跑 1 min 示例await broker.shutdown()
5.4 WebSocket 直播——網頁秒級訂閱
# 只改 URL
await client.connect('ws://broker.emqx.io:8083/mqtt')
5.5 SSL 上云——一參開啟加密
await client.connect('mqtts://broker.emqx.io:8883', cafile='/path/ca.pem')
6 三大常見坑：報錯 + 原因 + 急救方案
| # | 錯誤信息 | 根本原因 | 現成解藥 |
|---|---|---|---|
| 1 | NoDataException: No more data 頻繁出現 | 0.9.6 心跳線程 Bug，長時間空閑會自行斷鏈 | 升級至社區 fork amqtt==0.11.x；或手動 await client.ping() 每 30 s |
| 2 | publish 返回 0，但訂閱端收不到 | QoS0 不存儲、不重發；網絡閃斷即丟包 | 業務關鍵消息改用 qos=1 及以上 |
| 3 | OSError: [Errno 24] Too many open files | 單進程文件描述符默認 1024，萬連接瞬間打爆 | ulimit -n 65535 + 使用連接池復用客戶端 |
7 總結：遺產代碼安全駕駛指南
- hbmqtt 0.9.6 仍能穩定服役，但不再修 Bug；新項目請直接
pip install amqtt。 - 牢記“字節串、QoS、心跳”三大檢查點，可避開 90% 的坑。
- 若需 MQTT 5.0、共享訂閱等高級特性，轉向 gmqtt 或 paho-mqtt v5。
把這篇 cheatsheet 貼在工位，下次 IoT 需求來襲，你能在 15 分鐘內給出可擴展、可維護、不踩坑的異步 MQTT 方案。祝編碼愉快，消息永不丟！
如果你覺得文章還不錯，請大家 點贊、分享、留言 下，因為這將是我持續輸出更多優質文章的最強動力！
--Java版)
