后台任务和定时任务是后端系统的常见需求。本文讲解Celery分布式任务队列和APScheduler定时调度并给出实战案例。一、APScheduler定时任务python复制下载from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler BlockingScheduler() scheduler.scheduled_job(interval, minutes10) def job1(): print(每10分钟执行) scheduler.scheduled_job(CronTrigger(hour9, minute0)) def job2(): print(每天9点执行) scheduler.start()二、Flask/Django集成python复制下载from apscheduler.schedulers.background import BackgroundScheduler scheduler BackgroundScheduler() scheduler.add_job(job1, interval, minutes10) scheduler.start() # Flask关闭时 import atexit atexit.register(lambda: scheduler.shutdown())三、Celery基础Celery是分布式任务队列基于生产者-消费者模型。bash复制下载pip install celery redispython复制下载# tasks.py from celery import Celery app Celery(tasks, brokerredis://localhost:6379/0, backendredis://localhost:6379/1) app.task def add(a, b): return a b app.task(bindTrue, max_retries3) def send_email(self, to): try: # 发送邮件 pass except Exception as exc: raise self.retry(excexc, countdown60)四、调用任务python复制下载# 异步调用 result add.delay(1, 2) print(result.id) # 获取结果阻塞 print(result.get(timeout10)) # 定时调用 add.apply_async(args[1, 2], countdown60) # 指定时间 from datetime import datetime, timedelta add.apply_async(args[1, 2], etadatetime.now() timedelta(minutes5))五、Celery Beat定时任务python复制下载app.conf.beat_schedule { every-10-minutes: { task: tasks.add, schedule: 600.0, args: (1, 2), }, daily-report: { task: tasks.send_report, schedule: crontab(hour9, minute0), }, }启动bash复制下载celery -A tasks worker --loglevelinfo celery -A tasks beat --loglevelinfo六、任务路由与队列python复制下载app.conf.task_routes { tasks.send_email: {queue: email}, tasks.process_image: {queue: image}, } app.conf.task_queues { email: {exchange: email, routing_key: email}, image: {exchange: image, routing_key: image}, }bash复制下载celery -A tasks worker -Q email -c 4 celery -A tasks worker -Q image -c 2七、任务链与工作流python复制下载from celery import chain, group, chord # 链式 chain(add.s(1, 2), add.s(3), add.s(4))() # 并行 group(add.s(i, i) for i in range(10))() # chord并行后汇总 chord((add.s(i, i) for i in range(10)))(sum_task.s())八、监控Flowerbash复制下载pip install flower celery -A tasks flower --port5555访问http://localhost:5555查看任务状态。九、实战异步发送邮件python复制下载# tasks.py app.task(bindTrue, max_retries3) def send_email_task(self, to, subject, body): try: send_mail(to, subject, body) except SMTPException as exc: raise self.retry(excexc, countdown2 ** self.request.retries) # 视图中 def register(request): user create_user(request.POST) send_email_task.delay(user.email, 欢迎, 感谢注册) return JsonResponse({status: ok})十、总结APScheduler适合单机定时任务Celery适合分布式异步任务。掌握任务路由、重试、监控能构建可靠的后台系统。