Python基於mqtt異步編程主要用到asyncio及第三方庫hbmqtt,這裏主要介紹mqtt的異步發佈及遇到的一些問題。編程
hbmqtt安裝很簡單,pip hbmqtt install.服務器
mqtt服務器我使用的是mosquitto.session
1、主進程異步
主進程執行下面語句就實現了協程async
1 loop = asyncio.get_event_loop() 2 loop.run_until_complete(run())
首先是connect,而後publish,整個過程是一個協程異步編程
1 run(): #協程主函數 2 await connect() 3 while True: 4 try: 5 await publish() 6 except Exception as ce: 7 logger.error("Sender Error: %s" % ce)
2、Connect函數
這裏connect沒有使用自動重連機制connect(),而是單獨開一個協程執行掉線後自動重連_auto_reconnectoop
重連函數,每隔1s執行一次:ui
1 async def _auto_reconnect(client): 2 while True: 3 if not client.session.transitions.is_connected(): #若是已經鏈接上則不執行重連 4 try: 5 await client.reconnect() 6 except ConnectException: 7 pass 8 await asyncio.sleep(1)
鏈接函數,僅執行一次,並啓動重連函數協程url
1 async def connect(): 2 client = MQTTClient(config={"auto_reconnect": False}) #False時關閉自動重連 3 try: 4 await client.connect(url) 5 except ConnectException: 6 pass 7 asyncio.ensure_future(_auto_reconnect(client))
3、Publish:
發佈函數比較簡單,主要就是
1 await client.publish(topic, msg)
4、總結
這裏重點講爲何協程時不要自動重連,由於若是publish過程當中出現斷線,須要等待鏈接成功的event,若是容許自動重連"auto_reconnect": True,程序在publish程序等待信號不退出,沒法進入connect程序執行reconnect,這樣就永遠等不到信號,形成程序死等,相似死機。
若是不容許自動重連,單開一個協程執行重連操做,即便publish協程等待事件,重連協程會使這個事件響應,這樣就能夠繼續發佈。
在hbmqtt庫自帶例子中可能是先connect,而後publish,而後disconnect,以此循環,但主要考慮到鏈接後不主動斷開一提升程序效率,故沒有斷開操做。
這個問題的解決方式可能有點牽強,園友們有沒有遇到過相似的問題,在publish過程當中關閉mqtt服務器,再從新打開服務器,發佈任務可以繼續正常執行,若是有好的解決方式,還望不吝賜教。