使用python實(shí)現(xiàn)mqtt的發(fā)布和訂閱
需要安裝的python庫(kù)
使用python編寫程序進(jìn)行測(cè)試MQTT的發(fā)布和訂閱功能。首先要安裝:pip install paho-mqtt
測(cè)試發(fā)布(pub)
我的MQTT部署在阿里云的服務(wù)器上面,所以我在本機(jī)上編寫了python程序進(jìn)行測(cè)試。
然后在shell里面重新打開(kāi)一個(gè)終端,訂閱一個(gè)主題為“chat” mosquitto_sub -t chat
在本機(jī)上測(cè)試遠(yuǎn)程的MQTT的發(fā)布功能就是把自己作為一個(gè)發(fā)送信息的人,當(dāng)自己發(fā)送信息的時(shí)候,所有訂閱過(guò)該主題(topic)的對(duì)象都將收到自己發(fā)送的信息。
mqtt_client.py # encoding: utf-8 import paho.mqtt.client as mqtt HOST = "101.200.46.138" PORT = 1883 def test(): client = mqtt.Client() client.connect(HOST, PORT, 60) client.publish("chat","hello liefyuan",2) # 發(fā)布一個(gè)主題為'chat',內(nèi)容為‘hello liefyuan'的信息 client.loop_forever() if __name__ == '__main__': test()
發(fā)布/訂閱測(cè)試
# -*- coding: utf-8 -*- import paho.mqtt.client as mqtt MQTTHOST = "101.200.46.138" MQTTPORT = 1883 mqttClient = mqtt.Client() # 連接MQTT服務(wù)器 def on_mqtt_connect(): mqttClient.connect(MQTTHOST, MQTTPORT, 60) mqttClient.loop_start() # publish 消息 def on_publish(topic, payload, qos): mqttClient.publish(topic, payload, qos) # 消息處理函數(shù) def on_message_come(lient, userdata, msg): print(msg.topic + " " + ":" + str(msg.payload)) # subscribe 消息 def on_subscribe(): mqttClient.subscribe("/server", 1) mqttClient.on_message = on_message_come # 消息到來(lái)處理函數(shù) def main(): on_mqtt_connect() on_publish("/test/server", "Hello Python!", 1) on_subscribe() while True: pass if __name__ == '__main__': main()
注解函數(shù):
client.connect(self, host, port, keepalive, bind_address) client.publish(self, topic, payload, qos, retain) client.subscribe(self, topic, qos)
測(cè)試訂閱(sub)
在本機(jī)上編寫程序測(cè)試訂閱功能,就是讓自己的程序作為一個(gè)接收者,同一個(gè)主題沒(méi)有發(fā)布(pub)信息的時(shí)候,就自己一直等候。
# encoding: utf-8 import paho.mqtt.client as mqtt def on_connect(client, userdata, flags, rc): print("Connected with result code "+str(rc)) client.subscribe("chat") 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("www.liefyuan.top", 1883, 60) client.loop_forever()
總結(jié)
以上所述是小編給大家介紹的使用python實(shí)現(xiàn)mqtt的發(fā)布和訂閱,希望對(duì)大家有所幫助,如果大家有任何疑問(wèn)請(qǐng)給我留言,小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對(duì)腳本之家網(wǎng)站的支持!
如果你覺(jué)得本文對(duì)你有幫助,歡迎轉(zhuǎn)載,煩請(qǐng)注明出處,謝謝!
相關(guān)文章
利用Python中SocketServer 實(shí)現(xiàn)客戶端與服務(wù)器間非阻塞通信
本文主要介紹了利用Python中SocketServer 實(shí)現(xiàn)客戶端與服務(wù)器間非阻塞通信示例代碼,具有很好的參考價(jià)值,需要的朋友一起來(lái)看下吧2016-12-12python開(kāi)發(fā)之文件操作用法實(shí)例
這篇文章主要介紹了python開(kāi)發(fā)之文件操作用法,以實(shí)例形式較為詳細(xì)的分析了Python針對(duì)文件的路徑、文件名、后綴名等操作技巧,具有一定參考借鑒價(jià)值,需要的朋友可以參考下2015-11-11Python隨機(jī)生成數(shù)據(jù)后插入到PostgreSQL
本文主要介紹利用python的random庫(kù)生成隨機(jī)數(shù),然后插入到PostgreSQL數(shù)據(jù)庫(kù)中,有需要的可以參考學(xué)習(xí)。2016-07-07Django在視圖中使用表單并和數(shù)據(jù)庫(kù)進(jìn)行數(shù)據(jù)交互的實(shí)現(xiàn)
本文主要介紹了Django在視圖中使用表單并和數(shù)據(jù)庫(kù)進(jìn)行數(shù)據(jù)交互,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2022-07-07Python3.x檢查內(nèi)存可用大小的兩種實(shí)現(xiàn)
本文將介紹如何使用Python 3實(shí)現(xiàn)檢查L(zhǎng)inux服務(wù)器內(nèi)存可用大小的方法,包括使用Python標(biāo)準(zhǔn)庫(kù)實(shí)現(xiàn)和使用Linux命令實(shí)現(xiàn)兩種方式,感興趣可以了解一下2023-05-05Python3將數(shù)據(jù)保存為txt文件的方法
這篇文章主要介紹了Python3將數(shù)據(jù)保存為txt文件的方法,非常不錯(cuò),具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2019-09-09python3的UnicodeDecodeError解決方法
這篇文章主要介紹了python3的UnicodeDecodeError解決方法,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-12-12