pip install paho-mqtt
import threading
import paho.mqtt.client as mqtt
import time
HOST = "xxxxx" #emq服务器地址
PORT = 1883
class Mqtt_subscribe(threading.Thread):
"""
mqtt thread, 完成订阅功能
"""
def __init__(self, subtopic):
super(Mqtt_subscribe, self).__init__()
self.client_id = time.strftime(
'%Y%m%d%H%M%S', time.localtime(
time.time()))
self.client = mqtt.Client(self.client_id)
self.client.user_data_set(subtopic)
self.client.username_pw_set("admin", "public")
self.message = None
def run(self):
# ClientId不能重复,所以使用当前时间
# 必须设置,否则会返回「Connected with result code 4」
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
self.client.connect(HOST, PORT, 60)
self.client.loop_forever(timeout=60)
def on_connect(self, client, subtopic, flags, rc):
print("Connected with result code " + str(rc))
print("topic:" + subtopic)
client.subscribe(subtopic, 2)
def on_message(self, client, userdata, msg):
# print(msg.topic + " " + msg.payload.decode("utf-8"))
self.mess_age = msg.payload.decode("utf-8")
if 'test' in self.mess_age: #根据订阅内容判断,如果含有想要的值则会停止订阅
self.client.disconnect()
return self.mess_age
if __name__ == "__main__":
subtopic = "test"
t = Mqtt_subscribe(subtopic)
t.start()
import paho.mqtt.publish as publish
import time
def mqtt_Publish(topic,msg):
client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
publish.single(topic,msg,qos=0,
hostname="xxxx", #emq地址
port=1883,
client_id=client_id,
auth={'username': "admin", 'password': "public"}) #登录密码,官方是amdin,public
if __name__ == '__main__':
topic = 'test'
message = "this is a test"
mqtt_Publish(topic,message)
from testcase.test_04 import subscribe
topic = 'test'
mess = subscribe.Mqtt_subscribe(topic)
mess.start()
threads =[]
threads.append(mess)
for t in threads:
t.join(20) #超时20S不会再线程阻塞,如果没有接受到内容需要一直阻塞,则t.join()即可
a = mess.mess_age
print("这是变量信息",a)