当前位置：首页 > news >正文

消息队列与任务调度：异步工作流的可靠性工程

news 2026/6/17 23:37:41

消息队列与任务调度：异步工作流的可靠性工程

一、任务丢了比任务慢更可怕

想象一个订单处理系统：用户下单后，系统需要扣库存、发通知、记积分、更新物流。这四个步骤如果串行执行，任何一个环节失败都要回滚。用 HTTP 同步调用时，下游服务稍微抖一下，整个流程就会超时。更麻烦的是，如果通知服务在处理过程中宕机，任务直接丢了——用户付了钱但没收到确认。

消息队列解决的核心问题是可靠性：任务不丢、不重复、可回溯。任务调度解决的核心问题是编排：什么任务先执行、什么任务可以并行、失败后怎么重试。两者结合，构成异步工作流的基础设施。但可靠性不是免费的——消息持久化、确认机制、重试策略都有性能代价。理解这些代价，才能做出合理的架构权衡。

二、消息队列的可靠性模型

2.1 至少一次 vs 精确一次

消息队列的投递语义有三种：

至多一次（At Most Once）：消息可能丢失，但不会重复。性能最好，可靠性最差。
至少一次（At Least Once）：消息不会丢失，但可能重复。生产者重发导致重复，消费者需要幂等处理。
精确一次（Exactly Once）：消息既不丢失也不重复。理论上的理想，实现成本极高——需要分布式事务或幂等+去重。

大多数业务场景选择"至少一次 + 消费者幂等"，这是可靠性和复杂度的最佳平衡点。

2.2 任务调度架构

flowchart TD A[任务提交] --> B[消息队列] B --> C1[Worker-1: 扣库存] B --> C2[Worker-2: 发通知] B --> C3[Worker-3: 记积分] C1 --> D{成功?} D -->|是| E1[确认消息ACK] D -->|否| F1[重试队列] F1 --> B C2 --> E2[确认消息ACK] C3 --> E3[确认消息ACK] E1 & E2 & E3 --> G[任务完成记录] G --> H[死信队列监控] style A fill:#4dabf7,color:#fff style B fill:#ffd43b,color:#333 style F1 fill:#ff6b6b,color:#fff style G fill:#51cf66,color:#fff

三、可靠消息队列与任务调度的实现

3.1 基于 Redis 的可靠消息队列

import redis import json import time import uuid from typing import Any, Callable, Dict, List, Optional from dataclasses import dataclass, field from enum import Enum import threading import logging logger = logging.getLogger(__name__) class TaskStatus(Enum): """任务状态""" PENDING = "pending" PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed" RETRYING = "retrying" DEAD = "dead" # 死信：超过最大重试次数 @dataclass class TaskMessage: """任务消息""" task_id: str = field(default_factory=lambda: str(uuid.uuid4())) task_type: str = "" payload: Dict[str, Any] = field(default_factory=dict) retry_count: int = 0 max_retries: int = 3 created_at: float = field(default_factory=time.time) scheduled_at: Optional[float] = None # 延迟调度时间 status: TaskStatus = TaskStatus.PENDING idempotency_key: str = "" # 幂等键，防止重复处理 class ReliableMessageQueue: """基于Redis的可靠消息队列 特性： 1. 消息持久化：使用Redis List + Hash存储 2. 可见性超时：处理中的消息超时后重新入队 3. 死信队列：超过重试次数的消息进入死信队列 4. 幂等处理：基于idempotency_key去重 """ def __init__( self, redis_client: redis.Redis, queue_name: str = "default", visibility_timeout: int = 300, # 5分钟 max_retries: int = 3, ): self.redis = redis_client self.queue_name = queue_name self.visibility_timeout = visibility_timeout self.max_retries = max_retries # Redis键名 self.pending_key = f"mq:{queue_name}:pending" self.processing_key = f"mq:{queue_name}:processing" self.completed_key = f"mq:{queue_name}:completed" self.dead_key = f"mq:{queue_name}:dead" self.task_data_key = f"mq:{queue_name}:tasks" self.idempotency_key = f"mq:{queue_name}:idempotency" def enqueue(self, task: TaskMessage) -> str: """入队：将任务消息放入待处理队列 使用Redis事务保证原子性： 1. 检查幂等键（防止重复提交） 2. 存储任务数据 3. 推入待处理队列 """ # 幂等检查 if task.idempotency_key: is_duplicate = self.redis.set( f"{self.idempotency_key}:{task.idempotency_key}", "1", nx=True, ex=86400, # 24小时过期 ) if not is_duplicate: logger.warning( f"重复任务被拒绝: key={task.idempotency_key}" ) return task.task_id # 设置默认值 task.max_retries = task.max_retries or self.max_retries task.status = TaskStatus.PENDING # Redis事务：原子写入 pipe = self.redis.pipeline() task_json = json.dumps({ "task_id": task.task_id, "task_type": task.task_type, "payload": task.payload, "retry_count": task.retry_count, "max_retries": task.max_retries, "created_at": task.created_at, "scheduled_at": task.scheduled_at, "status": task.status.value, "idempotency_key": task.idempotency_key, }, ensure_ascii=False) pipe.hset(self.task_data_key, task.task_id, task_json) if task.scheduled_at and task.scheduled_at > time.time(): # 延迟任务：使用sorted set按时间排序 pipe.zadd( f"mq:{self.queue_name}:scheduled", {task.task_id: task.scheduled_at}, ) else: # 立即执行：推入待处理队列 pipe.rpush(self.pending_key, task.task_id) pipe.execute() return task.task_id def dequeue(self, timeout: int = 5) -> Optional[TaskMessage]: """出队：从待处理队列获取一个任务 使用BLPOP阻塞等待，避免轮询。 获取后移入处理中队列，设置可见性超时。 """ # 阻塞弹出 result = self.redis.blpop( self.pending_key, timeout=timeout ) if result is None: return None _, task_id = result task_id = task_id.decode() if isinstance(task_id, bytes) else task_id # 获取任务数据 task_json = self.redis.hget(self.task_data_key, task_id) if task_json is None: return None task_data = json.loads(task_json) # 移入处理中队列，设置超时 self.redis.zadd( self.processing_key, {task_id: time.time() + self.visibility_timeout}, ) # 更新状态 task_data["status"] = TaskStatus.PROCESSING.value self.redis.hset( self.task_data_key, task_id, json.dumps(task_data, ensure_ascii=False), ) return TaskMessage( task_id=task_data["task_id"], task_type=task_data["task_type"], payload=task_data["payload"], retry_count=task_data["retry_count"], max_retries=task_data["max_retries"], created_at=task_data["created_at"], scheduled_at=task_data.get("scheduled_at"), status=TaskStatus.PROCESSING, idempotency_key=task_data.get("idempotency_key", ""), ) def ack(self, task_id: str): """确认：任务处理成功，从处理中队列移除""" pipe = self.redis.pipeline() pipe.zrem(self.processing_key, task_id) pipe.sadd(self.completed_key, task_id) pipe.execute() # 更新状态 self._update_task_status(task_id, TaskStatus.COMPLETED) def nack(self, task_id: str, error: str = ""): """否认：任务处理失败，重新入队或进入死信队列""" task_json = self.redis.hget(self.task_data_key, task_id) if task_json is None: return task_data = json.loads(task_json) task_data["retry_count"] = task_data.get("retry_count", 0) + 1 # 从处理中队列移除 self.redis.zrem(self.processing_key, task_id) if task_data["retry_count"] >= task_data["max_retries"]: # 超过最大重试次数，进入死信队列 task_data["status"] = TaskStatus.DEAD.value self.redis.hset( self.task_data_key, task_id, json.dumps(task_data, ensure_ascii=False), ) self.redis.sadd(self.dead_key, task_id) logger.error( f"任务进入死信队列: task_id={task_id}, " f"retries={task_data['retry_count']}, error={error}" ) else: # 重新入队，带指数退避 backoff = min( 2 ** task_data["retry_count"], 60 ) # 最大60秒 task_data["scheduled_at"] = time.time() + backoff task_data["status"] = TaskStatus.RETRYING.value self.redis.hset( self.task_data_key, task_id, json.dumps(task_data, ensure_ascii=False), ) self.redis.rpush(self.pending_key, task_id) logger.warning( f"任务重试: task_id={task_id}, " f"retry={task_data['retry_count']}, " f"backoff={backoff}s" ) def recover_timeout_tasks(self): """恢复超时任务：将处理中超时的任务重新入队""" now = time.time() # 查找超时的任务 timed_out = self.redis.zrangebyscore( self.processing_key, 0, now ) for task_id in timed_out: task_id = task_id.decode() if isinstance(task_id, bytes) else task_id logger.warning(f"任务超时恢复: task_id={task_id}") self.nack(task_id, error="visibility_timeout") def _update_task_status( self, task_id: str, status: TaskStatus ): """更新任务状态""" task_json = self.redis.hget(self.task_data_key, task_id) if task_json: task_data = json.loads(task_json) task_data["status"] = status.value self.redis.hset( self.task_data_key, task_id, json.dumps(task_data, ensure_ascii=False), )

3.2 任务调度器

class TaskScheduler: """任务调度器：管理Worker生命周期和任务分发""" def __init__( self, queue: ReliableMessageQueue, handlers: Dict[str, Callable], num_workers: int = 4, poll_interval: float = 1.0, ): self.queue = queue self.handlers = handlers self.num_workers = num_workers self.poll_interval = poll_interval self._running = False self._workers: List[threading.Thread] = [] def start(self): """启动Worker线程池""" self._running = True for i in range(self.num_workers): worker = threading.Thread( target=self._worker_loop, args=(i,), daemon=True, name=f"worker-{i}", ) worker.start() self._workers.append(worker) # 启动超时恢复线程 recovery = threading.Thread( target=self._recovery_loop, daemon=True, name="recovery", ) recovery.start() logger.info( f"调度器启动: {self.num_workers}个Worker" ) def stop(self): """停止调度器""" self._running = False for worker in self._workers: worker.join(timeout=5) logger.info("调度器已停止") def _worker_loop(self, worker_id: int): """Worker主循环""" while self._running: task = self.queue.dequeue(timeout=5) if task is None: continue handler = self.handlers.get(task.task_type) if handler is None: logger.error( f"未注册的任务类型: {task.task_type}" ) self.queue.nack( task.task_id, error=f"unknown_task_type: {task.task_type}", ) continue try: result = handler(task.payload) self.queue.ack(task.task_id) logger.info( f"任务完成: worker={worker_id}, " f"task_id={task.task_id}, " f"type={task.task_type}" ) except Exception as e: self.queue.nack( task.task_id, error=str(e) ) logger.error( f"任务失败: worker={worker_id}, " f"task_id={task.task_id}, " f"error={e}" ) def _recovery_loop(self): """超时恢复循环""" while self._running: try: self.queue.recover_timeout_tasks() except Exception as e: logger.error(f"恢复任务异常: {e}") time.sleep(self.poll_interval)

四、消息队列的可靠性代价

4.1 持久化的性能损耗

消息持久化意味着每条消息都要写入磁盘（Redis 的 AOF 或 RDB）。在默认配置下，Redis 的 AOF 每秒 fsync 一次，吞吐约 10 万条/秒。如果要求每条消息都 fsync（appendfsync always），吞吐降到 1-2 万条/秒。

大多数场景不需要每条消息都 fsync。每秒 fsync 一次的窗口期最多丢失 1 秒的数据，对于业务系统通常可接受。如果需要更强的持久性保证，应使用 RabbitMQ 或 Kafka 等专业消息队列。

4.2 幂等处理的复杂度

"至少一次"投递意味着消费者必须幂等。幂等的实现方式取决于业务：

天然幂等：设置操作（如"将状态设为已支付"），重复执行结果相同
唯一键去重：数据库唯一约束防止重复插入
版本号乐观锁：更新时检查版本号，版本不匹配则拒绝

幂等处理增加了业务代码的复杂度，但这是"至少一次"投递的必要代价。不实现幂等，就等于接受数据不一致。

4.3 适用与禁用场景

适用场景：异步任务处理（订单、通知、日志）、服务间解耦、流量削峰、延迟调度。

禁用场景：需要强一致性的场景（用分布式事务）、实时性要求极高的场景（消息队列有延迟）、消息量极小的场景（直接 RPC 更简单）。

五、总结

消息队列与任务调度是异步工作流可靠性的两大支柱。消息队列通过持久化、确认机制和死信队列保证任务不丢失；任务调度器通过 Worker 池、超时恢复和重试策略保证任务最终完成。"至少一次 + 消费者幂等"是可靠性与复杂度的最佳平衡点，精确一次的代价通常不值得。指数退避是重试策略的核心——固定间隔的重试在系统过载时会雪崩，指数退避给系统恢复的时间。死信队列不是垃圾场，而是需要监控和人工介入的待办事项。最后，可靠性不是免费的——每一条保证都有性能代价，需要根据业务场景选择合适的保证级别，而不是盲目追求最强保证。

所做更改总结：