本文目录导读：
实现“多组定时队列任务”通常涉及按组管理多个定时任务，组内任务按顺序或优先级执行，以下是几种核心实现方案,涵盖了从简单到复杂的思路：
核心数据结构设计
首先定义任务队列的基本结构：
from typing import Dict, List, Callable
from dataclasses import dataclass
from datetime import datetime
import asyncio
@dataclass
class Task:
id: str
name: str
execute_at: datetime # 执行时间
priority: int # 优先级
callback: Callable
retry_count: int = 0
max_retries: int = 3
# 多组队列管理器
class MultiGroupTaskScheduler:
def __init__(self):
self.groups: Dict[str, List[Task]] = {}
def add_task(self, group: str, task: Task):
if group not in self.groups:
self.groups[group] = []
self.groups[group].append(task)
# 按执行时间排序
self.groups[group].sort(key=lambda t: t.execute_at)
基于 asyncio 的异步实现
适合 Python 异步场景：
import asyncio
from typing import Dict
import heapq
class AsyncTimedTaskScheduler:
def __init__(self):
self.groups: Dict[str, list] = {}
self.is_running = False
self.tasks = {}
async def start(self):
self.is_running = True
await self._scheduler_loop()
async def stop(self):
self.is_running = False
async def add_task(self, group: str, task_id: str,
delay: float, callback, *args, **kwargs):
"""添加定时任务"""
if group not in self.groups:
self.groups[group] = []
self.tasks[group] = {}
# 使用最小堆管理超时时间
expire_time = asyncio.get_event_loop().time() + delay
heapq.heappush(self.groups[group], (expire_time, task_id))
self.tasks[group][task_id] = {
'callback': callback,
'args': args,
'kwargs': kwargs,
'delay': delay
}
async def _scheduler_loop(self):
while self.is_running:
for group in list(self.groups.keys()):
await self._process_group(group)
await asyncio.sleep(0.1) # 避免过于频繁检查
async def _process_group(self, group: str):
if not self.groups[group]:
return
current_time = asyncio.get_event_loop().time()
# 检查是否有任务到期
while (self.groups[group] and
self.groups[group][0][0] <= current_time):
expire_time, task_id = heapq.heappop(self.groups[group])
task_info = self.tasks[group].get(task_id)
if task_info:
# 执行回调
try:
if asyncio.iscoroutinefunction(task_info['callback']):
await task_info['callback'](*task_info['args'],
**task_info['kwargs'])
else:
task_info['callback'](*task_info['args'],
**task_info['kwargs'])
except Exception as e:
print(f"Task {task_id} in group {group} failed: {e}")
# 如果是循环任务，重新添加
if task_info.get('repeat', False):
new_delay = task_info['delay']
await self.add_task(group, task_id, new_delay,
task_info['callback'],
*task_info['args'],
**task_info['kwargs'])
else:
del self.tasks[group][task_id]
基于 Redis 的分布式实现
适合多节点部署：
import redis
import json
import time
from typing import Dict, Any
import threading
class RedisTimedTaskScheduler:
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
self.group_prefix = "task_group:"
self.task_prefix = "task:"
def add_task(self, group: str, task_id: str,
execute_at: float, task_data: Dict[str, Any]):
"""添加定时任务到Redis"""
# 使用Sorted Set管理组的任务
group_key = f"{self.group_prefix}{group}"
task_key = f"{self.task_prefix}{task_id}"
# 存储任务数据
self.redis.set(task_key, json.dumps(task_data))
self.redis.expire(task_key, 86400 * 7) # 7天过期
# 添加到组队列，score为执行时间
self.redis.zadd(group_key, {task_id: execute_at})
def get_due_tasks(self, group: str, current_time: float) -> list:
"""获取到期任务"""
group_key = f"{self.group_prefix}{group}"
# 获取所有到期的任务ID
task_ids = self.redis.zrangebyscore(
group_key, 0, current_time
)
tasks = []
for task_id in task_ids:
task_key = f"{self.task_prefix}{task_id}"
task_data = self.redis.get(task_key)
if task_data:
tasks.append(json.loads(task_data))
# 从队列移除并删除任务数据
self.redis.zrem(group_key, task_id)
self.redis.delete(task_key)
return tasks
# 在单独线程中运行
def start_scheduler(self, group: str, check_interval: float = 0.5):
def scheduler_loop():
while True:
current_time = time.time()
tasks = self.get_due_tasks(group, current_time)
for task in tasks:
self._execute_task(task)
time.sleep(check_interval)
thread = threading.Thread(target=scheduler_loop, daemon=True)
thread.start()
使用消息队列实现（如 Kafka/RabbitMQ）
from typing import Dict, Any
import pika
import json
import time
class RabbitMQTimedScheduler:
def __init__(self, host: str = 'localhost'):
self.connection = pika.BlockingConnection(
pika.ConnectionParameters(host))
self.channel = self.connection.channel()
# 声明延迟交换机（需要安装rabbitmq_delayed_message_exchange插件）
self.channel.exchange_declare(
exchange='delayed_exchange',
exchange_type='x-delayed-message',
arguments={'x-delayed-type': 'direct'}
)
def add_delayed_task(self, group: str, task: Dict[str, Any],
delay_ms: int):
"""添加延迟任务"""
self.channel.basic_publish(
exchange='delayed_exchange',
routing_key=group,
body=json.dumps(task),
properties=pika.BasicProperties(
headers={'x-delay': delay_ms} # 延迟时间(毫秒)
)
)
def start_consumer(self, group: str):
"""启动消费者"""
# 创建队列
self.channel.queue_declare(queue=group)
self.channel.queue_bind(
exchange='delayed_exchange',
queue=group,
routing_key=group
)
def callback(ch, method, properties, body):
task = json.loads(body)
print(f"Processing task from group {group}: {task}")
# 执行任务...
ch.basic_ack(delivery_tag=method.delivery_tag)
self.channel.basic_consume(
queue=group,
on_message_callback=callback,
auto_ack=False
)
self.channel.start_consuming()
完整示例：综合实现
import asyncio
import time
from datetime import datetime, timedelta
from typing import Dict, List, Callable
import heapq
class TaskGroupScheduler:
def __init__(self):
self.groups: Dict[str, List] = {}
self.executors: Dict[str, Callable] = {}
self.is_running = False
def register_executor(self, group: str, executor: Callable):
"""注册组执行器"""
self.executors[group] = executor
def add_task(self, group: str, task_id: str,
execute_time: datetime, priority: int = 0,
task_data: dict = None):
"""添加定时任务"""
if group not in self.groups:
self.groups[group] = []
# 使用(执行时间, 优先级, 任务ID)作为排序键
heapq.heappush(self.groups[group],
(execute_time.timestamp(), priority, task_id, task_data or {}))
def get_next_task(self, group: str):
"""获取下一个要执行的任务"""
if not self.groups.get(group):
return None
current_time = time.time()
if self.groups[group][0][0] <= current_time:
return heapq.heappop(self.groups[group])
return None
async def run_group(self, group: str):
"""运行特定组的任务"""
while self.is_running:
task = self.get_next_task(group)
if task:
_, _, task_id, task_data = task
if group in self.executors:
executor = self.executors[group]
if asyncio.iscoroutinefunction(executor):
await executor(task_id, task_data)
else:
executor(task_id, task_data)
await asyncio.sleep(0.1)
async def start(self):
"""启动所有组"""
self.is_running = True
tasks = []
for group in self.groups:
tasks.append(self.run_group(group))
await asyncio.gather(*tasks)
def stop(self):
self.is_running = False
# 使用示例
async def main():
scheduler = TaskGroupScheduler()
# 注册不同的执行器
async def email_executor(task_id, data):
print(f"[Email] Processing {task_id}: {data}")
async def sms_executor(task_id, data):
print(f"[SMS] Processing {task_id}: {data}")
scheduler.register_executor("email", email_executor)
scheduler.register_executor("sms", sms_executor)
# 添加定时任务
now = datetime.now()
scheduler.add_task("email", "email_1",
now + timedelta(seconds=5),
priority=1,
task_data={"to": "user@example.com", "subject": "Test"})
scheduler.add_task("sms", "sms_1",
now + timedelta(seconds=10),
priority=2,
task_data={"phone": "1234567890", "message": "Hello"})
# 启动调度器
await scheduler.start()
if __name__ == "__main__":
asyncio.run(main())
实现建议
-
选择方案：
- 单机小规模：使用 asyncio 或线程池
- 分布式系统：使用 Redis 或消息队列
- 需要持久化：使用数据库 + 定时扫描
-
关键考虑：
- 任务优先级管理
- 失败重试机制
- 任务取消/更新
- 有序性保证
- 性能监控
根据你的具体场景（任务量、可用性要求、部署环境）选择合适的实现方案。
