init

.gitignore

@@ -0,0 +1,48 @@
+
+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/
+/src/main/resources/log/
+
+### Pycharm ###
+/venv/
+/log/
+/build/
+__pycache__/
+*.pyc
+
+### custom ###
+/@AutomationLog.txt
+node_modules/
+dist/
+/logs/

mqtt_publisher/__init__.py

@@ -0,0 +1,3 @@
+# @description: 
+# @author: licanglong
+# @date: 2025/11/12 11:57

mqtt_publisher/mqtt_client.log

@@ -0,0 +1,11 @@
+2025-11-12 15:30:00,266 [INFO] [🚀] 客户端 mqtt_client_3c7376b0 启动成功
+2025-11-12 15:30:00,275 [INFO] [❤️] 发送心跳: mqtt_client_3c7376b0 心跳 1762932600
+2025-11-12 15:30:00,312 [INFO] [✅] 成功连接到 MQTT Broker: 117.72.147.109:18830
+2025-11-12 15:30:10,276 [INFO] [❤️] 发送心跳: mqtt_client_3c7376b0 心跳 1762932610
+2025-11-12 15:30:20,277 [INFO] [❤️] 发送心跳: mqtt_client_3c7376b0 心跳 1762932620
+2025-11-12 15:30:30,277 [INFO] [❤️] 发送心跳: mqtt_client_3c7376b0 心跳 1762932630
+2025-11-12 15:30:40,278 [INFO] [❤️] 发送心跳: mqtt_client_3c7376b0 心跳 1762932640
+2025-11-12 15:30:50,281 [INFO] [❤️] 发送心跳: mqtt_client_3c7376b0 心跳 1762932650
+2025-11-12 15:31:00,281 [INFO] [❤️] 发送心跳: mqtt_client_3c7376b0 心跳 1762932660
+2025-11-12 15:31:10,282 [INFO] [❤️] 发送心跳: mqtt_client_3c7376b0 心跳 1762932670
+2025-11-12 15:31:15,539 [INFO] 🛑 手动终止连接

mqtt_publisher/scheduler.py

@@ -0,0 +1,49 @@
+# @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 = "scheduler"
+
+online_workers = {}
+
+
+def on_message(client, userdata, msg):
+    data = json.loads(msg.payload.decode())
+    worker_id = data["client_id"]
+    status = data.get("status", "unknown")
+
+    if status == "online":
+        online_workers[worker_id] = time.time()
+    elif status == "offline":
+        online_workers.pop(worker_id, None)
+
+
+def send_task(client, task_data):
+    now = time.time()
+    active_workers = [w for w, t in online_workers.items() if now - t < 30]
+    if not active_workers:
+        print("暂无在线worker，等待中...")
+        return
+    worker = active_workers[0]
+    topic = f"task/{worker}"
+    client.publish(topic, json.dumps(task_data))
+    print(f"任务已发送给 {worker}: {task_data}")
+
+
+# ✅ 新写法：兼容 paho-mqtt 2.x
+client = mqtt.Client(client_id=client_id, callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
+client.on_message = on_message
+client.connect(broker, port, 60)
+
+client.subscribe("worker/status/+")
+client.loop_start()
+
+while True:
+    send_task(client, {"task_id": int(time.time()), "action": "compute"})
+    time.sleep(10)

mqtt_subscriber/__init__.py

@@ -0,0 +1,3 @@
+# @description: 
+# @author: licanglong
+# @date: 2025/11/12 11:57

mqtt_subscriber/worker.py

@@ -0,0 +1,38 @@
+# @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)