python3 mqtt 客户端以及服务端

pip3 install paho-mqtt

#!/usr/bin/env python
#coding=utf- import json
import sys
import os
import paho.mqtt.client as mqtt
import timesys.path.append(os.path.abspath(os.path.dirname(__file__) + '/' + '..'))
sys.path.append("..")TASK_TOPIC = 'test' # 客户端发布消息主题client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))client = mqtt.Client(client_id, transport='tcp')client.connect("127.0.0.1", , ) # 此处端口默认为1883，通信端口期keepalive默认60
client.loop_start()def clicent_main(message: str):
"""
客户端发布消息
:param message: 消息主体
:return:
"""
time_now = time.strftime('%Y-%m-%d %H-%M-%S', time.localtime(time.time()))
payload = {"msg": "%s" % message, "data": "%s" % time_now}
# publish(主题：Topic; 消息内容)
client.publish(TASK_TOPIC, json.dumps(payload, ensure_ascii=False))
print("Successful send message!")
return Trueif __name__ == '__main__':
msg = "我是一条测试数据！"
clicent_main(msg)

#!/usr/bin/env python
#coding=utf- import json
import sys
import os
import time
import paho.mqtt.client as mqtt
sys.path.append(os.path.abspath(os.path.dirname(__file__) + '/' + '..'))
sys.path.append("..")REPORT_TOPIC = 'test' # 主题def on_connect(client, userdata, flags, rc):
print('connected to mqtt with resurt code ', rc)
client.subscribe(REPORT_TOPIC) # 订阅主题def on_message(client, userdata, msg):
"""
接收客户端发送的消息
:param client: 连接信息
:param userdata:
:param msg: 客户端返回的消息
:return:
"""
print("Start server!")
payload = json.loads(msg.payload.decode('utf-8'))
print(payload)def server_conenet(client):
client.on_connect = on_connect # 启用订阅模式
client.on_message = on_message # 接收消息
client.connect("127.0.0.1", , ) # 链接
# client.loop_start() # 以start方式运行，需要启动一个守护线程，让服务端运行，否则会随主线程死亡
client.loop_forever() # 以forever方式阻塞运行。def server_stop(client):
client.loop_stop() # 停止服务端
sys.exit()def server_main():
client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
client = mqtt.Client(client_id, transport='tcp')
server_conenet(client)if __name__ == '__main__':
# 启动监听
server_main()