Python\u5b9e\u4f8b\u9898<\/p>\n
\u9898\u76ee<\/p>\n
\u95ee\u9898\u63cf\u8ff0<\/p>\n
\u8c03\u5ea6\u4e2d\u5fc3（Server）：<\/p>\n
\u5de5\u4f5c\u8282\u70b9（Worker）：<\/p>\n
\u5ba2\u6237\u7aef（Client）：<\/p>\n
\u7cfb\u7edf\u67b6\u6784<\/p>\n
\u5173\u952e\u5b9e\u73b0\u601d\u8def<\/p>\n
\u6838\u5fc3\u4ee3\u7801\u6846\u67b6（\u8c03\u5ea6\u4e2d\u5fc3\u90e8\u5206）<\/p>\n
\u96be\u70b9\u5206\u6790<\/p>\n
\u6269\u5c55\u65b9\u5411<\/p>\n<\/p>\n
\u5206\u5e03\u5f0f\u4efb\u52a1\u8c03\u5ea6\u7cfb\u7edf<\/p>\n
\u8bbe\u8ba1\u4e00\u4e2a\u7b80\u5316\u7248\u5206\u5e03\u5f0f\u4efb\u52a1\u8c03\u5ea6\u7cfb\u7edf，\u5305\u542b\u4ee5\u4e0b\u7ec4\u4ef6：<\/p>\n
+—————-+ +—————-+ +—————-+
\n| \u5ba2\u6237\u7aef |<—>| \u8c03\u5ea6\u4e2d\u5fc3 |<—>| \u5de5\u4f5c\u8282\u70b91 |
\n+—————-+ +—————-+ +—————-+
\n ^
\n |
\n+—————-+ +—————-+ +—————-+
\n| \u5ba2\u6237\u7aef |<—>| \u8c03\u5ea6\u4e2d\u5fc3 |<—>| \u5de5\u4f5c\u8282\u70b92 |
\n+—————-+ +—————-+ +—————-+<\/p>\n
import socket
\nimport json
\nimport threading
\nimport time
\nimport sqlite3
\nimport queue
\nfrom typing import Dict, List, Tuple, Optional<\/p>\n
# \u901a\u4fe1\u534f\u8bae\u5b9a\u4e49
\nclass Protocol:
\n # \u6d88\u606f\u7c7b\u578b
\n TYPE_TASK_SUBMIT = "TASK_SUBMIT" # \u63d0\u4ea4\u4efb\u52a1
\n TYPE_TASK_ASSIGN = "TASK_ASSIGN" # \u5206\u914d\u4efb\u52a1
\n TYPE_TASK_RESULT = "TASK_RESULT" # \u8fd4\u56de\u7ed3\u679c
\n TYPE_WORKER_HEARTBEAT = "HEARTBEAT" # \u5de5\u4f5c\u8282\u70b9\u5fc3\u8df3
\n TYPE_WORKER_REGISTER = "REGISTER" # \u5de5\u4f5c\u8282\u70b9\u6ce8\u518c<\/p>\n
@staticmethod
\n def pack_message(msg_type: str, data: dict) -> bytes:
\n """\u6253\u5305\u6d88\u606f\u4e3aJSON\u683c\u5f0f"""
\n message = {"type": msg_type, "data": data, "timestamp": time.time()}
\n return json.dumps(message).encode()<\/p>\n
@staticmethod
\n def unpack_message(msg: bytes) -> dict:
\n """\u89e3\u6790JSON\u6d88\u606f"""
\n try:
\n return json.loads(msg.decode())
\n except:
\n return {"type": "ERROR", "data": "Invalid message format"}<\/p>\n
# \u4efb\u52a1\u7ba1\u7406\u7c7b
\nclass TaskManager:
\n def __init__(self, db_path="task_system.db"):
\n self.task_queue = queue.PriorityQueue() # \u4f18\u5148\u7ea7\u961f\u5217，\u6570\u5b57\u8d8a\u5c0f\u4f18\u5148\u7ea7\u8d8a\u9ad8
\n self.task_status = {} # task_id: {"status": "", "worker": "", "result": ""}
\n self.worker_nodes = {} # worker_id: {"address": "", "last_heartbeat": 0, "busy": False}
\n self.db_path = db_path
\n self._init_database()<\/p>\n
def _init_database(self):
\n """\u521d\u59cb\u5316\u6570\u636e\u5e93"""
\n conn = sqlite3.connect(self.db_path)
\n cursor = conn.cursor()
\n cursor.execute('''
\n CREATE TABLE IF NOT EXISTS tasks (
\n task_id TEXT PRIMARY KEY,
\n name TEXT,
\n priority INTEGER,
\n status TEXT,
\n input_data TEXT,
\n result TEXT,
\n worker_id TEXT,
\n create_time REAL,
\n update_time REAL
\n )
\n ''')
\n conn.commit()
\n conn.close()<\/p>\n
def add_task(self, task_id: str, name: str, priority: int, data: dict) -> bool:
\n """\u6dfb\u52a0\u4efb\u52a1\u5230\u961f\u5217"""
\n if task_id in self.task_status:
\n return False<\/p>\n
self.task_status[task_id] = {
\n "status": "PENDING",
\n "worker": "",
\n "result": None,
\n "create_time": time.time()
\n }
\n self.task_queue.put((priority, task_id, name, data))<\/p>\n
# \u5b58\u50a8\u5230\u6570\u636e\u5e93
\n conn = sqlite3.connect(self.db_path)
\n cursor = conn.cursor()
\n cursor.execute(
\n "INSERT INTO tasks (task_id, name, priority, status, input_data, create_time, update_time) "
\n "VALUES (?, ?, ?, ?, ?, ?, ?)",
\n (task_id, name, priority, "PENDING", json.dumps(data), time.time(), time.time())
\n )
\n conn.commit()
\n conn.close()
\n return True<\/p>\n
def get_available_worker(self) -> Optional[str]:
\n """\u83b7\u53d6\u53ef\u7528\u5de5\u4f5c\u8282\u70b9"""
\n current_time = time.time()
\n available_workers = []<\/p>\n
# \u8fc7\u6ee4\u5b58\u6d3b\u4e14\u7a7a\u95f2\u7684\u8282\u70b9（10\u79d2\u5185\u6709\u5fc3\u8df3）
\n for worker_id, info in self.worker_nodes.items():
\n if (not info["busy"]) and (current_time – info["last_heartbeat"] < 10):
\n available_workers.append(worker_id)<\/p>\n
if available_workers:
\n # \u7b80\u5355\u8f6e\u8be2\u6216\u9009\u62e9\u8d1f\u8f7d\u6700\u4f4e\u7684\u8282\u70b9
\n return available_workers[0]
\n return None<\/p>\n
def assign_task(self) -> Optional[Tuple[str, str, dict]]:
\n """\u5206\u914d\u4efb\u52a1\u7ed9\u5de5\u4f5c\u8282\u70b9"""
\n if self.task_queue.empty():
\n return None<\/p>\n
# \u83b7\u53d6\u6700\u9ad8\u4f18\u5148\u7ea7\u4efb\u52a1
\n priority, task_id, task_name, task_data = self.task_queue.get()<\/p>\n
# \u68c0\u67e5\u4efb\u52a1\u72b6\u6001
\n if self.task_status[task_id]["status"] != "PENDING":
\n return self.assign_task() # \u4efb\u52a1\u72b6\u6001\u5df2\u53d8\u66f4，\u91cd\u65b0\u5206\u914d<\/p>\n
# \u83b7\u53d6\u53ef\u7528\u5de5\u4f5c\u8282\u70b9
\n worker_id = self.get_available_worker()
\n if not worker_id:
\n # \u65e0\u53ef\u7528\u8282\u70b9，\u5c06\u4efb\u52a1\u653e\u56de\u961f\u5217
\n self.task_queue.put((priority, task_id, task_name, task_data))
\n return None<\/p>\n
# \u66f4\u65b0\u4efb\u52a1\u72b6\u6001
\n self.task_status[task_id] = {
\n **self.task_status[task_id],
\n "status": "RUNNING",
\n "worker": worker_id,
\n "update_time": time.time()
\n }<\/p>\n
# \u66f4\u65b0\u5de5\u4f5c\u8282\u70b9\u72b6\u6001
\n self.worker_nodes[worker_id]["busy"] = True<\/p>\n
# \u66f4\u65b0\u6570\u636e\u5e93
\n conn = sqlite3.connect(self.db_path)
\n cursor = conn.cursor()
\n cursor.execute(
\n "UPDATE tasks SET status = ?, worker_id = ?, update_time = ? WHERE task_id = ?",
\n ("RUNNING", worker_id, time.time(), task_id)
\n )
\n conn.commit()
\n conn.close()<\/p>\n
return (worker_id, task_id, task_data)<\/p>\n
def process_result(self, task_id: str, result: dict, worker_id: str) -> bool:
\n """\u5904\u7406\u4efb\u52a1\u7ed3\u679c"""
\n if task_id not in self.task_status:
\n return False<\/p>\n
# \u66f4\u65b0\u4efb\u52a1\u72b6\u6001
\n self.task_status[task_id] = {
\n **self.task_status[task_id],
\n "status": "COMPLETED",
\n "result": result,
\n "update_time": time.time()
\n }<\/p>\n
# \u91ca\u653e\u5de5\u4f5c\u8282\u70b9
\n if worker_id in self.worker_nodes:
\n self.worker_nodes[worker_id]["busy"] = False<\/p>\n
# \u5b58\u50a8\u7ed3\u679c\u5230\u6570\u636e\u5e93
\n conn = sqlite3.connect(self.db_path)
\n cursor = conn.cursor()
\n cursor.execute(
\n "UPDATE tasks SET status = ?, result = ?, worker_id = ?, update_time = ? WHERE task_id = ?",
\n ("COMPLETED", json.dumps(result), "", time.time(), task_id)
\n )
\n conn.commit()
\n conn.close()
\n return True<\/p>\n
def register_worker(self, worker_id: str, address: str) -> bool:
\n """\u6ce8\u518c\u5de5\u4f5c\u8282\u70b9"""
\n if worker_id in self.worker_nodes:
\n return False<\/p>\n
self.worker_nodes[worker_id] = {
\n "address": address,
\n "last_heartbeat": time.time(),
\n "busy": False
\n }
\n return True<\/p>\n
def update_heartbeat(self, worker_id: str) -> bool:
\n """\u66f4\u65b0\u5de5\u4f5c\u8282\u70b9\u5fc3\u8df3"""
\n if worker_id in self.worker_nodes:
\n self.worker_nodes[worker_id]["last_heartbeat"] = time.time()
\n return True
\n return False<\/p>\n
# \u8c03\u5ea6\u4e2d\u5fc3\u670d\u52a1\u5668\u7c7b
\nclass SchedulerServer:
\n def __init__(self, host="0.0.0.0", port=9999):
\n self.host = host
\n self.port = port
\n self.task_manager = TaskManager()
\n self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
\n self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
\n self.running = False
\n self.clients = {} # client_id: socket<\/p>\n
def start(self):
\n """\u542f\u52a8\u670d\u52a1\u5668"""
\n self.server_socket.bind((self.host, self.port))
\n self.server_socket.listen(10)
\n self.running = True
\n print(f"\u8c03\u5ea6\u4e2d\u5fc3\u542f\u52a8\u6210\u529f，\u76d1\u542c\u5730\u5740: {self.host}:{self.port}")<\/p>\n
# \u542f\u52a8\u5fc3\u8df3\u68c0\u6d4b\u7ebf\u7a0b
\n heartbeat_thread = threading.Thread(target=self._heartbeat_check, daemon=True)
\n heartbeat_thread.start()<\/p>\n
# \u63a5\u53d7\u5ba2\u6237\u7aef\u8fde\u63a5
\n while self.running:
\n try:
\n client_socket, address = self.server_socket.accept()
\n print(f"\u65b0\u8fde\u63a5: {address}")
\n client_thread = threading.Thread(
\n target=self._handle_client,
\n args=(client_socket, address),
\n daemon=True
\n )
\n client_thread.start()
\n except Exception as e:
\n print(f"\u63a5\u53d7\u8fde\u63a5\u5931\u8d25: {e}")
\n if self.running:
\n time.sleep(1)<\/p>\n
def _handle_client(self, client_socket: socket.socket, address: tuple):
\n """\u5904\u7406\u5ba2\u6237\u7aef\u8fde\u63a5"""
\n client_id = f"{address[0]}:{address[1]}"
\n self.clients[client_id] = client_socket<\/p>\n
try:
\n while True:
\n # \u63a5\u6536\u6d88\u606f
\n data = client_socket.recv(4096)
\n if not data:
\n break<\/p>\n
# \u89e3\u6790\u6d88\u606f
\n message = Protocol.unpack_message(data)
\n msg_type = message.get("type", "UNKNOWN")
\n msg_data = message.get("data", {})
\n print(f"\u6536\u5230\u6d88\u606f [{client_id}]: {msg_type}")<\/p>\n
# \u5904\u7406\u4e0d\u540c\u7c7b\u578b\u6d88\u606f
\n if msg_type == Protocol.TYPE_WORKER_REGISTER:
\n worker_id = msg_data.get("worker_id")
\n if self.task_manager.register_worker(worker_id, client_id):
\n response = Protocol.pack_message(
\n "REGISTER_SUCCESS",
\n {"message": "Worker registered successfully"}
\n )
\n else:
\n response = Protocol.pack_message(
\n "REGISTER_FAILED",
\n {"message": "Worker ID already exists"}
\n )
\n client_socket.send(response)<\/p>\n
elif msg_type == Protocol.TYPE_WORKER_HEARTBEAT:
\n worker_id = msg_data.get("worker_id")
\n if self.task_manager.update_heartbeat(worker_id):
\n # \u5c1d\u8bd5\u5206\u914d\u4efb\u52a1
\n task = self.task_manager.assign_task()
\n if task:
\n worker_id, task_id, task_data = task
\n response = Protocol.pack_message(
\n Protocol.TYPE_TASK_ASSIGN,
\n {"task_id": task_id, "data": task_data}
\n )
\n else:
\n response = Protocol.pack_message(
\n "NO_TASK_ASSIGN",
\n {"message": "No task available"}
\n )
\n else:
\n response = Protocol.pack_message(
\n "HEARTBEAT_FAILED",
\n {"message": "Worker not registered"}
\n )
\n client_socket.send(response)<\/p>\n
elif msg_type == Protocol.TYPE_TASK_SUBMIT:
\n task_id = msg_data.get("task_id")
\n task_name = msg_data.get("task_name", "Unknown Task")
\n priority = msg_data.get("priority", 5) # \u9ed8\u8ba4\u4e3a\u4e2d\u7b49\u4f18\u5148\u7ea7
\n task_data = msg_data.get("data", {})<\/p>\n
if self.task_manager.add_task(task_id, task_name, priority, task_data):
\n response = Protocol.pack_message(
\n "TASK_SUBMIT_SUCCESS",
\n {"task_id": task_id, "message": "Task submitted"}
\n )
\n else:
\n response = Protocol.pack_message(
\n "TASK_SUBMIT_FAILED",
\n {"message": "Task ID already exists"}
\n )
\n client_socket.send(response)<\/p>\n
elif msg_type == Protocol.TYPE_TASK_RESULT:
\n task_id = msg_data.get("task_id")
\n result = msg_data.get("result", {})
\n worker_id = msg_data.get("worker_id")<\/p>\n
if self.task_manager.process_result(task_id, result, worker_id):
\n response = Protocol.pack_message(
\n "RESULT_RECEIVED",
\n {"task_id": task_id, "message": "Result received"}
\n )
\n else:
\n response = Protocol.pack_message(
\n "RESULT_FAILED",
\n {"message": "Invalid task ID"}
\n )
\n client_socket.send(response)<\/p>\n
else:
\n response = Protocol.pack_message(
\n "UNKNOWN_MESSAGE",
\n {"message": "Unknown message type"}
\n )
\n client_socket.send(response)<\/p>\n
except Exception as e:
\n print(f"\u5ba2\u6237\u7aef\u5904\u7406\u9519\u8bef [{client_id}]: {e}")
\n finally:
\n if client_id in self.clients:
\n del self.clients[client_id]
\n client_socket.close()
\n print(f"\u8fde\u63a5\u5173\u95ed: {client_id}")<\/p>\n
def _heartbeat_check(self):
\n """\u5b9a\u671f\u68c0\u67e5\u5de5\u4f5c\u8282\u70b9\u5fc3\u8df3"""
\n while self.running:
\n current_time = time.time()
\n for worker_id, info in list(self.worker_nodes.items()):
\n if current_time – info["last_heartbeat"] > 30:
\n # \u8282\u70b9\u8d85\u65f6，\u6807\u8bb0\u4e3a\u79bb\u7ebf
\n print(f"\u5de5\u4f5c\u8282\u70b9 {worker_id} \u8d85\u65f6，\u6807\u8bb0\u4e3a\u79bb\u7ebf")
\n info["busy"] = False # \u91ca\u653e\u53ef\u80fd\u6b63\u5728\u5904\u7406\u7684\u4efb\u52a1
\n time.sleep(10) # \u6bcf10\u79d2\u68c0\u67e5\u4e00\u6b21<\/p>\n
def stop(self):
\n """\u505c\u6b62\u670d\u52a1\u5668"""
\n self.running = False
\n if self.server_socket:
\n self.server_socket.close()
\n print("\u8c03\u5ea6\u4e2d\u5fc3\u5df2\u505c\u6b62")<\/p>\n
# \u4f7f\u7528\u793a\u4f8b
\nif __name__ == "__main__":
\n server = SchedulerServer(port=9999)
\n try:
\n server.start()
\n except KeyboardInterrupt:
\n print("\u63a5\u6536\u5230\u505c\u6b62\u4fe1\u53f7")
\n finally:
\n server.stop()<\/p>\n
\u6587\u7ae0\u6d4f\u89c8\u9605\u8bfb243\u6b21\uff0c\u70b9\u8d5e3\u6b21\uff0c\u6536\u85cf3\u6b21\u3002\u5206\u5e03\u5f0f\u4efb\u52a1\u8c03\u5ea6\u7cfb\u7edf\u3002<\/p>\n","protected":false},"author":2,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[81,664,4125,190],"topic":[],"class_list":["post-44313","post","type-post","status-publish","format-standard","hentry","category-server","tag-python","tag-664","tag-4125","tag-190"],"yoast_head":"\n