怎样实现管理多组定时队列任务

本文目录导读：

1. 核心数据结构设计
2. 基于 asyncio 的异步实现
3. 基于 Redis 的分布式实现
4. 使用消息队列实现（如 Kafka/RabbitMQ）
5. 完整示例：综合实现
6. 实现建议

实现“多组定时队列任务”通常涉及按组管理多个定时任务，组内任务按顺序或优先级执行，以下是几种核心实现方案,涵盖了从简单到复杂的思路：

核心数据结构设计

首先定义任务队列的基本结构：

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())

实现建议

1. 选择方案：

   • 单机小规模：使用 asyncio 或线程池
   • 分布式系统：使用 Redis 或消息队列
   • 需要持久化：使用数据库 + 定时扫描

2. 关键考虑：

   • 任务优先级管理
   • 失败重试机制
   • 任务取消/更新
   • 有序性保证
   • 性能监控

根据你的具体场景（任务量、可用性要求、部署环境）选择合适的实现方案。