在尝试将AI推理能力从中心化的云端下沉到分布式边缘节点时，我们面临的第一个棘手问题并非模型本身，而是网络拓扑与服务生命周期的不确定性。边缘节点，无论是工厂车间的工控机还是零售店的服务器，它们的网络环境复杂多变，且随时可能因为断电、维护或网络抖动而离线。在这种环境下，依赖传统的、基于DNS或重型注册中心（如Consul、Etcd）的服务发现机制，不仅部署成本高昂，其中心化的心跳检测模式也难以应对边缘网络的延迟和不稳定性。
我们的目标是构建一个轻量级、去中心化、且对网络瞬断有良好容忍度的服务发现与请求路由层。这个系统需要能够动态感知在线的Hugging Face推理工作节点，并将客户端的推理请求高效地分发给它们。技术选型直接指向了ZeroMQ，它并非一个消息队列，而是一个用于构建高性能、分布式网络应用的套接字库。其内置的通信模式，特别是ROUTER/DEALER和PUB/SUB，为解决这个问题提供了几乎完美的构件。
架构构想：信标与动态路由
我们的核心构想是放弃中心化的注册表，转而采用一种“信标广播”模型。
推理工作节点 (Worker): 每个边缘节点上运行一个Worker进程。它加载一个Hugging Face模型，并通过一个
DEALER套接字准备接收推理任务。同时，它会启动一个后台线程，通过PUB套接字周期性地向网络广播一个“我还活着”的信标消息。这个信标包含了它的唯一ID和用于接收任务的DEALER地址。服务代理/路由器 (Broker): 在一个相对稳定的节点（可以是云端，也可以是区域汇聚节点）上，我们部署一个Broker。这个Broker是整个架构的神经中枢，但它自身不维护持久化的状态。
- 它通过一个
SUB套接字订阅所有Worker发出的信标。当收到信标时，就在内存中更新一个可用Worker列表，记录Worker的ID、地址以及最后一次收到信标的时间戳。 - 它通过一个
ROUTER套接字接收来自客户端的请求。当请求到达时，它从可用Worker列表中选择一个（例如，通过轮询），并将请求连同原始客户端的身份信息一起转发给该Worker的DEALER套接字。 - Broker会有一个清理线程，定期检查Worker列表，将长时间未发送信标的Worker（即已失联的）从中移除。
- 它通过一个
客户端 (Client): 客户端完全不关心Worker的存在。它只知道Broker的地址，并通过一个
REQ套接字将所有请求发送给Broker。
这种架构的优势在于：
- 解耦: Worker的上线和下线对客户端是透明的。Worker只需要广播自身存在，无需向任何中心节点“注册”。
- 韧性: Broker是无状态的（Worker列表是动态构建的）。即使Broker重启，只要Worker还在广播信标，服务就能在短时间内自动恢复。
- 轻量: 整个通信层完全由ZeroMQ构件组成，无需额外部署和维护数据库或注册中心集群。
下面是这个架构的交互流程图。
sequenceDiagram
participant C as Client
participant B as Broker
participant W1 as Worker 1
participant W2 as Worker 2
W1->>+B: PUB Heartbeat (ID: W1, Addr: tcp://..:5560)
W2->>+B: PUB Heartbeat (ID: W2, Addr: tcp://..:5561)
B-->>-W1: (Broker SUB socket receives, updates internal registry)
B-->>-W2: (Broker SUB socket receives, updates internal registry)
loop Heartbeat Loop
W1-->>B: PUB Heartbeat
W2-->>B: PUB Heartbeat
end
C->>+B: REQ (Inference Request)
B-->>-C: (ROUTER socket receives from Client)
Note over B: Selects an available worker (e.g., W1)
B->>+W1: Forward Request (via DEALER to Worker's DEALER)
W1-->>-B: (Processes request with Transformers model)
W1->>+B: Reply (Inference Result)
B-->>-W1: (ROUTER socket receives from Worker)
B->>+C: Forward Reply to original client
C-->>-B: (REQ socket receives final result)
核心实现：推理工作节点 (Worker)
Worker是整个系统的基础单元。它需要承担两项职责：执行模型推理和广播自身状态。我们使用multiprocessing来将这两项任务分离，避免信标广播阻塞推理任务。
worker.py
import zmq
import time
import uuid
import logging
import os
import threading
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
# --- 配置 ---
HEARTBEAT_INTERVAL_S = 2
WORKER_TTL_S = 5 # Worker存活时间，超过此时间未收到心跳则认为离线
MODEL_NAME = "distilbert-base-uncased-finetuned-sst-2-english"
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
# --- 日志设置 ---
logging.basicConfig(level=LOG_LEVEL, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger("HF-Worker")
class InferenceWorker:
def __init__(self, worker_id, broker_pub_url, broker_dealer_url):
"""
初始化一个推理工作节点.
:param worker_id: 工作节点的唯一标识符.
:param broker_pub_url: Broker用于发布信标的SUB套接字地址.
:param broker_dealer_url: Broker用于路由任务的ROUTER套接字地址.
"""
self.worker_id = worker_id
self.broker_pub_url = broker_pub_url
self.broker_dealer_url = broker_dealer_url
self.context = zmq.Context.instance()
# 用于接收任务的DEALER套接字
self.task_socket = self.context.socket(zmq.DEALER)
# 必须设置身份，Broker的ROUTER才能知道回复给谁
self.task_socket.setsockopt_string(zmq.IDENTITY, self.worker_id)
self.task_socket.connect(self.broker_dealer_url)
# 用于发送心跳的PUB套接字
self.heartbeat_socket = self.context.socket(zmq.PUB)
self.heartbeat_socket.connect(self.broker_pub_url)
self.model = None
self._stop_event = threading.Event()
logger.info(f"Worker {self.worker_id} initialized.")
def _load_model(self):
"""
加载Hugging Face模型。这是一个耗时操作，应在启动时完成。
在真实项目中，这里可能还包括模型下载、缓存检查等逻辑。
"""
logger.info(f"Loading model: {MODEL_NAME}...")
try:
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME)
self.model = pipeline("sentiment-analysis", model=model, tokenizer=tokenizer)
logger.info("Model loaded successfully.")
except Exception as e:
logger.error(f"Failed to load model: {e}")
raise
def _send_heartbeat(self):
"""
通过心跳线程周期性地发送信标。
信标内容格式为 'WORKER_READY|worker_id'.
"""
while not self._stop_event.is_set():
try:
# 信标消息包含了状态和ID
# 在更复杂的场景中，可以包含更多信息，如当前负载、GPU内存等
heartbeat_msg = f"WORKER_READY|{self.worker_id}"
self.heartbeat_socket.send_string(heartbeat_msg)
logger.debug(f"Sent heartbeat: {heartbeat_msg}")
except zmq.ZMQError as e:
logger.error(f"ZMQ error sending heartbeat: {e}")
time.sleep(HEARTBEAT_INTERVAL_S)
def start(self):
"""
启动Worker，包括加载模型、启动心跳线程和主任务循环。
"""
self._load_model()
heartbeat_thread = threading.Thread(target=self._send_heartbeat, daemon=True)
heartbeat_thread.start()
logger.info(f"Heartbeat thread started. Broadcasting to {self.broker_pub_url}")
logger.info(f"Worker {self.worker_id} ready to receive tasks from {self.broker_dealer_url}")
poller = zmq.Poller()
poller.register(self.task_socket, zmq.POLLIN)
try:
while not self._stop_event.is_set():
socks = dict(poller.poll(timeout=1000))
if self.task_socket in socks:
# DEALER套接字接收到的消息是多部分的：[client_identity, empty_frame, request_payload]
# 这是ROUTER转发过来的格式
parts = self.task_socket.recv_multipart()
if len(parts) != 3:
logger.warning(f"Received malformed message with {len(parts)} parts.")
continue
client_addr, _, request_data = parts
request_text = request_data.decode('utf-8')
logger.info(f"Received task from client {client_addr.hex()}: '{request_text}'")
# 执行模型推理
try:
result = self.model(request_text)
response_data = str(result).encode('utf-8')
except Exception as e:
logger.error(f"Inference failed for text '{request_text}': {e}")
response_data = b"ERROR:Inference failed"
# 将结果发回，同样是多部分消息，ROUTER会根据client_addr找到原始客户端
self.task_socket.send_multipart([client_addr, b'', response_data])
except KeyboardInterrupt:
logger.info("Shutting down worker...")
finally:
self.stop()
def stop(self):
self._stop_event.set()
self.task_socket.close()
self.heartbeat_socket.close()
# self.context.term() # 在多线程/进程中，由主进程管理Context生命周期
logger.info(f"Worker {self.worker_id} stopped.")
if __name__ == "__main__":
# 在生产环境中，这些URL应该是可配置的
BROKER_PUB_URL = os.environ.get("BROKER_PUB_URL", "tcp://127.0.0.1:5555")
BROKER_DEALER_URL = os.environ.get("BROKER_ROUTER_URL", "tcp://127.0.0.1:5556")
worker_id = f"worker-{uuid.uuid4()}"
worker = InferenceWorker(worker_id, BROKER_PUB_URL, BROKER_DEALER_URL)
worker.start()
这里的关键点：
-
DEALER套接字的IDENTITY: 必须为DEALER套接字设置一个唯一的身份标识。当ROUTER套接字（在Broker中）收到这个Worker的回复时，它才知道消息来自哪个Worker。虽然在这个架构中Broker主要关心客户端身份，但设置Worker身份是良好实践。 - 心跳内容: 信标消息
WORKER_READY|{self.worker_id}简单明了。Broker可以据此解析出Worker的状态和ID。 - 多线程: 推理和心跳在不同线程中，确保了即使推理任务耗时较长，心跳也能准时发出，避免被Broker误判为离线。
核心实现：服务代理 (Broker)
Broker是整个系统的粘合剂。它不执行业务逻辑，只负责维护动态路由表和转发消息。
broker.py
import zmq
import time
import logging
import os
from collections import deque
# --- 配置 ---
HEARTBEAT_INTERVAL_S = 2
WORKER_TTL_S = 5 # Worker存活时间，超过此时间未收到心跳则认为离线
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
# --- 日志设置 ---
logging.basicConfig(level=LOG_LEVEL, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger("Broker")
class ServiceBroker:
def __init__(self, frontend_url, backend_url, monitor_url):
"""
初始化服务代理。
:param frontend_url: 客户端连接的地址 (ROUTER).
:param backend_url: Worker连接的地址 (DEALER).
:param monitor_url: 监听Worker心跳的地址 (SUB).
"""
self.frontend_url = frontend_url
self.backend_url = backend_url
self.monitor_url = monitor_url
self.context = zmq.Context.instance()
# 面向客户端的ROUTER套接字
self.frontend = self.context.socket(zmq.ROUTER)
self.frontend.bind(self.frontend_url)
# 面向Worker的DEALER套接字
self.backend = self.context.socket(zmq.DEALER)
self.backend.bind(self.backend_url)
# 监听心跳的SUB套接字
self.monitor = self.context.socket(zmq.SUB)
self.monitor.bind(self.monitor_url)
self.monitor.setsockopt_string(zmq.SUBSCRIBE, "WORKER_READY") # 只订阅特定前缀的消息
# 内存中的服务注册表
self.workers = {} # key: worker_id, value: expiry_time
self.worker_queue = deque() # 用于实现简单的轮询负载均衡
logger.info(f"Broker initialized. Frontend: {frontend_url}, Backend: {backend_url}, Monitor: {monitor_url}")
def _purge_workers(self):
"""
清理超时的、不活跃的Worker。
这是一个在生产环境中至关重要的容错机制。
"""
now = time.time()
expired_workers = [wid for wid, expiry in self.workers.items() if now > expiry]
for worker_id in expired_workers:
logger.warning(f"Worker {worker_id} expired. Removing from registry.")
del self.workers[worker_id]
# 从队列中移除可能需要更高效的数据结构，但对于中等规模，这是可接受的
if worker_id in self.worker_queue:
# deque不支持高效的remove，这里是简化处理
new_queue = deque([w for w in self.worker_queue if w != worker_id])
self.worker_queue = new_queue
def _handle_heartbeat(self):
"""处理来自Worker的心跳消息。"""
try:
message = self.monitor.recv_string(flags=zmq.NOBLOCK)
parts = message.split('|')
if len(parts) == 2 and parts[0] == "WORKER_READY":
worker_id = parts[1]
if worker_id not in self.workers:
logger.info(f"New worker registered: {worker_id}")
self.worker_queue.append(worker_id)
# 更新Worker的过期时间
self.workers[worker_id] = time.time() + WORKER_TTL_S
else:
logger.warning(f"Received invalid heartbeat message: {message}")
except zmq.Again:
pass # 没有心跳消息，正常
def run(self):
"""
启动Broker的主事件循环。
"""
poller = zmq.Poller()
# 同时监听来自客户端和Worker的请求
poller.register(self.frontend, zmq.POLLIN)
poller.register(self.backend, zmq.POLLIN)
last_purge_time = time.time()
logger.info("Broker is running...")
try:
while True:
# 首先处理心跳和清理，确保路由表是最新的
self._handle_heartbeat()
if time.time() - last_purge_time > 1.0:
self._purge_workers()
last_purge_time = time.time()
# 使用短暂的超时，以便循环可以定期处理心跳和清理
socks = dict(poller.poll(timeout=100))
if self.frontend in socks:
# 收到客户端请求
# 消息格式: [client_identity, empty_frame, request_payload]
request = self.frontend.recv_multipart()
if not self.worker_queue:
logger.error("No available workers to handle request. Dropping.")
# 在真实场景中，可以返回一个错误信息给客户端
# self.frontend.send_multipart([request[0], b'', b'ERROR:No workers available'])
continue
# 轮询选择一个Worker
worker_id = self.worker_queue.popleft()
self.worker_queue.append(worker_id)
# 将客户端请求原封不动地转发给选定的Worker
# 消息格式: [worker_identity, client_identity, empty_frame, request_payload]
# DEALER 会自动添加自己的身份，但我们需要手动将worker身份和客户端请求组合
# 修正：ROUTER-DEALER模式下，DEALER会负载均衡，ROUTER-ROUTER需要手动路由
# 这里使用ROUTER-DEALER，所以后端直接发送即可
# 关键在于Worker的DEALER连接到了Broker的DEALER，形成了一个队列
# Oops, 这里的架构理解有偏差。应该是ROUTER-ROUTER，或者ROUTER-DEALER
# 如果后端是DEALER，它会自动负载均衡到连接它的PEERs(Workers)
# 我们的设计是Broker主动选择worker，所以后端应该是ROUTER
# 这里为了代码简洁，先用DEALER的自动负载均衡，但这意味着失去了主动选择Worker的能力
# 让我们坚持最初的设计：Broker主动选择。为此，后端必须是ROUTER，Worker必须是DEALER
# 但是Broker的DEALER连接到Worker的DEALER是行不通的。
# 正确模式：Client(REQ) -> Broker(ROUTER) -> Broker(DEALER) -> Worker(ROUTER)
# 让我们简化一下，Broker使用一个后端ROUTER，它直接连接到所有Worker的DEALER
# 但Worker地址是动态的。
# 最经典的负载均衡模式是：
# Client(REQ) -> Broker(ROUTER)
# Worker(REQ) -> Broker(DEALER)
# 这是请求-回复模式的负载均衡。
# 我们当前的设计是完全异步的。
# Client(REQ) -> Broker(ROUTER)
# Broker(ROUTER) -> Worker(DEALER)
# Worker(DEALER) -> Broker(ROUTER) -> Client(REQ)
# 这里后端用DEALER更合适，因为它能自动轮询分发。
logger.info(f"Routing request from {request[0].hex()} to worker {worker_id}")
# 转发给后端，DEALER套接字会自动处理负载均衡
# 消息格式：[client_addr, b'', request_data]
self.backend.send_multipart(request)
if self.backend in socks:
# 收到Worker的回复，转发回客户端
reply = self.backend.recv_multipart()
self.frontend.send_multipart(reply)
except KeyboardInterrupt:
logger.info("Shutting down broker...")
finally:
self.frontend.close()
self.backend.close()
self.monitor.close()
self.context.term()
if __name__ == "__main__":
# 生产环境中应从配置中读取
FRONTEND_URL = os.environ.get("BROKER_FRONTEND_URL", "tcp://*:5557")
# 这个是Worker连接的地址，也是Client连接的地址，ROUTER/DEALER应该在同一个套接字上。
# 让我们修正架构：一个ROUTER处理所有事情。
# Client(REQ) -> Broker(ROUTER) <- Worker(DEALER)
# 这样更简单。
# 我们重新设计Broker
# ------------------ 修正后的Broker实现 ------------------
# 这个版本更符合ZeroMQ的模式，也更健壮
logger.info("--- Starting Corrected Broker Implementation ---")
class CorrectedServiceBroker:
def __init__(self, frontend_url):
self.frontend_url = frontend_url
self.context = zmq.Context.instance()
# 一个ROUTER套接字处理客户端和Worker的所有通信
self.socket = self.context.socket(zmq.ROUTER)
self.socket.bind(frontend_url)
# 心跳监听部分保持不变
self.monitor = self.context.socket(zmq.SUB)
self.monitor.bind("tcp://*:5555") # Hardcode for simplicity
self.monitor.setsockopt_string(zmq.SUBSCRIBE, "WORKER_READY")
self.workers = {}
self.worker_queue = deque()
logger.info(f"Corrected Broker initialized. Listening on {frontend_url}")
def _purge_workers(self):
# (同上)
now = time.time()
expired_workers = [wid for wid, expiry in self.workers.items() if now > expiry]
for worker_id in expired_workers:
logger.warning(f"Worker {worker_id} expired.")
if worker_id in self.workers: del self.workers[worker_id]
if worker_id in self.worker_queue:
# Inefficient removal from deque
self.worker_queue = deque([w for w in self.worker_queue if w != worker_id])
def _handle_heartbeat(self):
# (同上)
try:
msg = self.monitor.recv_string(flags=zmq.NOBLOCK)
_, worker_id = msg.split('|')
if worker_id not in self.workers:
logger.info(f"Worker {worker_id} registered.")
self.worker_queue.append(worker_id)
self.workers[worker_id] = time.time() + WORKER_TTL_S
except zmq.Again:
pass
def run(self):
last_purge_time = time.time()
poller = zmq.Poller()
poller.register(self.socket, zmq.POLLIN)
while True:
self._handle_heartbeat()
if time.time() - last_purge_time > 1.0:
self._purge_workers()
last_purge_time = time.time()
socks = dict(poller.poll(timeout=100))
if self.socket in socks:
# ROUTER收到的消息格式: [identity, empty_frame, payload]
msg = self.socket.recv_multipart()
sender_id = msg[0]
# 区分消息是来自Worker还是Client
# 我们约定Worker的身份以"worker-"开头
if sender_id.decode('utf-8') in self.workers:
# 这是来自Worker的回复: [worker_id, empty, client_id, empty, payload]
# Worker的DEALER发回时，格式是 [client_id, empty, payload]
# ROUTER收到后，会加上worker的id: [worker_id, client_id, empty, payload]
client_id = msg[1]
logger.debug(f"Received reply from worker {sender_id.decode()}, forwarding to client {client_id.hex()}")
# 转发给客户端：[client_id, empty, payload]
self.socket.send_multipart([client_id, b'', msg[3]])
else:
# 这是来自客户端的请求: [client_id, empty, payload]
if not self.worker_queue:
logger.error("No workers available. Replying with error.")
self.socket.send_multipart([sender_id, b'', b'ERROR:NO_WORKERS'])
continue
# 轮询选择一个Worker
worker_id = self.worker_queue.popleft()
self.worker_queue.append(worker_id)
logger.info(f"Forwarding request from client {sender_id.hex()} to worker {worker_id}")
# 格式: [worker_id, empty, client_id, empty, payload]
# ROUTER socket发送时，第一帧是目标地址
forward_msg = [worker_id.encode('utf-8'), b'', sender_id, b'', msg[2]]
self.socket.send_multipart(forward_msg)
# 运行修正后的Broker
broker = CorrectedServiceBroker("tcp://*:5556")
broker.run()在实现过程中，我发现最初的Broker设计（一个ROUTER和一个DEALER）对于主动路由选择来说过于复杂且不符合ZeroMQ的最佳实践。一个单一的ROUTER套接字处理所有入站和出站流量是更清晰、更强大的模式。
修正后的CorrectedServiceBroker逻辑如下：
- 统一入口: 只使用一个
ROUTER套接字。无论是客户端还是Worker，都连接到这个地址。 - 身份识别: 当
ROUTER收到消息时，第一帧是发送方的身份。我们通过检查这个身份是否在我们的已知Worker列表中，来判断消息是来自客户端的请求，还是来自Worker的回复。 - 显式路由: 当收到客户端请求时，我们从
worker_queue中选择一个Worker ID，并构建一个多部分消息[worker_id, b'', client_id, b'', payload]发送出去。ROUTER套接字会确保这个消息只被发送到ID为worker_id的对端。 - 回复转发: 当收到Worker的回复
[worker_id, client_id, b'', payload]时，我们提取出client_id，然后将[client_id, b'', payload]发回，ROUTER会将其正确地路由回原始客户端。
这个修正后的模型更加健壮，并且完全体现了ROUTER套接字的强大路由能力。
客户端实现
客户端的实现非常简单，它只需要知道Broker的地址，然后使用REQ套接字发送请求并等待回复。
client.py
import zmq
import time
import sys
import logging
# --- 日志设置 ---
logging.basicConfig(level="INFO", format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger("Client")
def run_client(broker_url, text_to_analyze):
context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.connect(broker_url)
# REQ套接字需要设置超时，防止因Broker或Worker无响应而永久阻塞
socket.setsockopt(zmq.RCVTIMEO, 5000) # 5秒超时
socket.setsockopt(zmq.LINGER, 0) # 退出时立即关闭套接字
logger.info(f"Sending request: '{text_to_analyze}'")
try:
socket.send_string(text_to_analyze)
reply = socket.recv_string()
logger.info(f"Received reply: {reply}")
except zmq.Again:
logger.error("Request timed out. No response from broker.")
except Exception as e:
logger.error(f"An error occurred: {e}")
finally:
socket.close()
context.term()
if __name__ == "__main__":
if len(sys.argv) < 2:
print(f"Usage: python {sys.argv[0]} \"some text to analyze\"")
sys.exit(1)
BROKER_URL = "tcp://127.0.0.1:5556"
text = sys.argv[1]
run_client(BROKER_URL, text)客户端的关键在于设置了RCVTIMEO，这是一个生产环境中必须的配置，防止网络问题导致客户端永久挂起。
单元测试思路
对这样一个分布式系统进行单元测试的重点在于模拟各个组件的行为。
- 测试Broker的Worker管理:
- 启动Broker。
- 模拟一个Worker发送心跳。断言
broker.workers和broker.worker_queue被正确更新。 - 停止模拟心跳。等待
WORKER_TTL_S秒后，断言该Worker被从注册表中清除。
- 测试Broker的路由逻辑:
- 启动Broker，并模拟一个已注册的Worker。
- 模拟一个Client发送请求到Broker的
ROUTER套接字。 - 使用一个测试套接字捕获Broker发出的消息，断言该消息的目标地址是已注册的Worker ID，并且消息内容包含了原始客户端的ID。
- 测试Worker的健壮性:
- 发送格式错误的任务给Worker，断言其不会崩溃并能正确处理异常。
- 测试模型加载失败的场景，断言Worker进程会以错误状态退出或记录严重错误。
当前方案的局限性与未来展望
这个基于ZeroMQ的实现提供了一个非常轻量和高效的动态服务发现与负载均衡机制，但它并非银弹。在投入生产前，必须清楚它的边界：
Broker单点故障 (SPOF): 尽管Broker本身是无状态的，但它仍然是通信路径上的单点。如果Broker宕机，整个服务将中断。可以通过主备模式（例如，使用VRRP）或构建一个更复杂的、去中心化的Broker集群来解决，但这会显著增加复杂性。
简单的负载均衡策略: 当前使用的是简单的轮询。它没有考虑每个Worker的实际负载（例如，CPU/GPU使用率、当前任务队列长度）。一个更优化的策略是让Worker在心跳信标中附带自身的负载信息，Broker据此进行加权轮询或选择最空闲的节点。
服务质量 (QoS): 该方案没有内置的重试机制。如果Broker将任务发给一个Worker后，该Worker在处理过程中崩溃，任务就会丢失。需要实现应用级别的确认和重试逻辑，例如客户端在超时后重新发送请求，Broker需要有能力处理重复请求（幂等性）。
安全性: 当前所有通信都是明文的。ZeroMQ通过CurveZMQ提供了强大的端到端加密和认证机制，在生产环境中部署时，必须启用它来防止窃听和未经授权的节点接入。
尽管存在这些局限，但该架构展示了如何利用ZeroMQ的强大能力，为资源受限、网络不稳定的边缘计算场景量身打造一个高性能的分布式服务层。未来的迭代将集中在解决上述问题，逐步增强其在生产环境中的鲁棒性和可管理性。
