Python中如何实现一个执行定时任务（Crontab）和任务队列（Taskqueue）的小工具

我来写一个结合APScheduler和Celery的定时任务与队列工具。

# task_manager.py
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from celery import Celery
import time

# Celery配置
celery_app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',  # Redis作为消息代理
    backend='redis://localhost:6379/0'   # Redis存储结果
)

# 定义Celery任务
@celery_app.task
def process_task(task_id, data):
    """异步任务处理函数"""
    print(f"[{datetime.now()}] 开始处理任务 {task_id}: {data}")
    time.sleep(2)  # 模拟耗时操作
    result = f"任务 {task_id} 完成: {data.upper()}"
    print(f"[{datetime.now()}] {result}")
    return result

# APScheduler定时器
scheduler = BackgroundScheduler()

def schedule_task():
    """定时触发Celery任务"""
    task_data = {"message": f"定时任务于 {datetime.now()}"}
    result = process_task.delay("cron_job", task_data)
    print(f"已提交定时任务，任务ID: {result.id}")

# 添加定时任务
scheduler.add_job(
    schedule_task,
    'cron',
    second='*/30',  # 每30秒执行一次
    id='my_cron_job'
)

def start_scheduler():
    """启动定时调度器"""
    scheduler.start()
    print("定时调度器已启动")
    
    try:
        # 保持主线程运行
        while True:
            time.sleep(1)
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        print("定时调度器已停止")

if __name__ == "__main__":
    # 启动Celery worker: celery -A task_manager worker --loglevel=info
    # 启动定时调度器
    start_scheduler()

# 手动提交任务的示例
# submit_task.py
from task_manager import process_task

# 手动提交异步任务
if __name__ == "__main__":
    # 提交即时任务
    task1 = process_task.delay("manual_1", {"action": "发送邮件"})
    task2 = process_task.delay("manual_2", {"action": "生成报告"})
    
    print(f"任务1 ID: {task1.id}")
    print(f"任务2 ID: {task2.id}")
    
    # 获取结果（可选）
    # result = task1.get(timeout=10)
    # print(f"任务1结果: {result}")

这个方案用APScheduler处理cron定时，Celery管理任务队列。启动需要Redis服务，运行celery -A task_manager worker --loglevel=info启动worker，再跑task_manager.py启动定时器。

简单说就是APScheduler管定时，Celery管队列。