# -*- coding:utf-8 -*-
from multiprocessing import Process
import paho.mqtt.publish as publish
from Data.data import *
import paho.mqtt.subscribe as subscribe
from common.yaml_common_method import *
import time,json,yaml,os
"""基礎資料yaml獲取"""
data = yaml_method_read("data")
host = data["host"]#IP地址
prot = data["port"]#埠
pub_url = data["pub_url"]+"/register"#發送
sub_url = data["sub_url"]+"/register/response"#訂閱
"""主題請求引數yaml獲取"""
publish_yaml = yaml_method_read("publish")
publish_msg = publish_yaml["upload_service"]
client_id = time.strftime('mq:test:%Y%m%d%H%M%S',time.localtime(time.time()))
class Mqtt_method_info():
def mqtt_publish(self):
#這里是發送的函式
msg = publish_msg
print("------------------------------Send_sub------------------------------")
print(pub_url)
print("----------------------------Send_message----------------------------")
print(msg)
msg = json.dumps(msg)
publish.single(pub_url, msg, qos=1, hostname=host, port=prot, client_id=client_id)
print("------------------------------End_send------------------------------")
def mqtt_subscribe(self):
#這里是訂閱的函式
print("--------------------------Subscribe_topics--------------------------")
print(sub_url)
msg = subscribe.simple(sub_url, qos=1, hostname=host, port=prot, client_id=client_id)
msg_payload = json.loads(msg.payload.decode("utf-8"))
print("---------------------------Return_message----------------------------")
print(msg_payload)
yaml_method_write("mqtt_return_data",msg_payload)
def implement(self):
pub = Process(target=Mqtt_method_info().mqtt_publish)#發送執行緒
sub = Process(target=Mqtt_method_info().mqtt_subscribe)#訂閱執行緒
sub.start()
time.sleep(1)
pub.start()
sub.join()
print("執行完畢")
if __name__ == '__main__':
Mqtt_method_info().implement()這里講一下python-mqtt測驗腳本：
我們會使用到python的paho-mqtt庫，同樣pip安裝下就行，失敗的話大部分是需要用鏡像站的情況
這份代碼分為yaml檔案讀取發送的資料，這里只有操作的方法，大家如果需要使用可以把發送、訂閱的函式中修改，只運行方法，然后自己再通過各種檔案管理資料
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/302863.html
標籤：其他
上一篇：S3C2440驅動開發(八)
下一篇：高級驅動——（驅動所有按鍵）