mqtt_example/mqtt_subscriber/worker.py

# @description: 
# @author: licanglong
# @date: 2025/11/12 15:38
import json
import time

import paho.mqtt.client as mqtt

broker = "117.72.147.109"
port = 18830
client_id = "worker-01"

client = mqtt.Client(client_id=client_id, callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
client.will_set(f"worker/status/{client_id}",
                payload=json.dumps({"status": "offline", "client_id": client_id}),
                retain=True)


def on_message(client, userdata, msg):
    data = json.loads(msg.payload.decode())
    print(f"收到任务: {data}")
    # 模拟任务执行
    time.sleep(3)
    client.publish("task/ack", json.dumps({"task_id": data["task_id"], "worker": client_id, "result": "done"}))


client.on_message = on_message
client.connect(broker, port, 60)

client.subscribe(f"task/{client_id}")
client.loop_start()

# 定时发送心跳
while True:
    client.publish(f"worker/status/{client_id}",
                   json.dumps({"status": "online", "client_id": client_id, "timestamp": int(time.time())}),
                   retain=True)
    time.sleep(10)