介绍
🧩 什么是“定时任务”？
定时任务，就是按照设定的时间间隔或时间点自动执行某些操作。比如：
• 每天早上8点发通知
• 每隔10秒采集一次数据
• 每小时清理一次缓存相关使用
✅ 最简单的方式：while True + time.sleep()
import timedef job():print("执行任务")while True:job()time.sleep(10) # 每10秒执行一次
✅ 优点：
• 写法简单，不需要任何依赖
• 控制力强
❌ 缺点：
• 会阻塞当前线程
• 精度差（任务执行时间会影响间隔）
• 没有“精确到几点几分”的调度能力
✅ 进阶方案：使用 schedule 库
pip install schedule
import schedule
import timedef job():print("每隔5秒执行一次")schedule.every(5).seconds.do(job)while True:schedule.run_pending()time.sleep(1)>>>>>>>>>
schedule.every().day.at("10:30").do(job) 每天10:30
schedule.every().monday.at("09:00").do(job) 每周一上午9点
• schedule.run_pending() 会检查：是否有任务应该执行
• 如果是，就执行对应的 job()
• 然后 time.sleep(1) 让主线程休息 1 秒，再循环检查
⏱ 所以：任务是每5秒一次，不是“休息6秒”。sleep(1) 是用于“轮询检测任务是否该执行”，并不会影响任务周期本身。
✅ 优点：
• 语法优雅、简单
• 支持 every().minutes, every().day.at(“10:00”) 这种写法
• 适合小型任务管理
✅ 更强大的方案：使用 APScheduler
from apscheduler.schedulers.blocking import BlockingSchedulerdef job():print("每5秒执行一次")scheduler = BlockingScheduler() # 创建一个阻塞型调度器实例
scheduler.add_job(job, 'interval', seconds=5) # 每隔5秒调度一次 job 函数
scheduler.start() # 启动调度器（这个会阻塞主线程）
内部原理确实 本质上就是一个“定时任务调度器 + 任务轮询器”。
APScheduler 提供了多种调度器，比如：
• BlockingScheduler: 启动后会阻塞主线程，适合简单脚本
• BackgroundScheduler: 在后台启动，不阻塞主线程
• AsyncIOScheduler, TornadoScheduler, TwistedScheduler: 分别适配不同异步框架
| 类型 | 含义 |
|---|---|
| interval | 固定时间间隔 |
| cron | 类似 crontab 表达式，支持精确到秒 |
| date | 只运行一次，指定时间点 |
调用 start() 后，它会启动一个内部的 循环调度线程（或事件循环）：
• 持续维护一个 “任务执行计划表”（内部是优先队列）
• 每一轮 tick（可能是 0.5~1 秒级别），检查是否有任务到了执行时间
• 有就执行（用线程池或进程池执行）
| 特性 | 说明 |
|---|---|
| 支持多种调度类型 | interval / cron / date |
| 有任务注册中心 | 管理所有待运行的任务 |
| 有时间轮询机制 | 类似事件循环，周期性检查是否“触发” |
| 可以并发执行 | 默认使用线程池 |
| 可持久化任务 | 支持 SQLite、Redis 等存储后恢复 |
基于asyncio + 自定义时间
优点：
• 自由度高：可以灵活计算下一次执行时间。
• 无需额外依赖，原生 asyncio 就能跑。
• 跟 Redis 结合很好，可以做跨进程/跨机器任务协调。
缺点：
• 手动管理定时逻辑（get_next_run_time + asyncio.sleep()）。
• 多任务可能不好管理，比如暂停/重启某个 job。
• 不支持 cron 表达式等复杂调度。
import asyncio
import logging
import os
from datetime import datetime, timedeltaimport redis.asyncio as redislogging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)class ScheduleService:def __init__(self):self.redis_subscribe_key = "demo:subscription"self.exist_subscribe_key = "demo:exist_push"self.push_interval_seconds = 600 # 推送间隔时间self.fixed_times = ["08:00", "12:00", "18:00"] # 固定调度时间列表def get_next_run_time(self):now_time = datetime.now()today_str = now_time.strftime("%Y-%m-%d")# 今日的所有调度时间点run_times = [datetime.strptime(f"{today_str} {t}", "%Y-%m-%d %H:%M") for t in self.fixed_times]future_times = [t for t in run_times if t > now_time]if future_times:return min(future_times)else:# 如果今天已经过了所有调度时间，则返回明天的第一个时间点next_day = (now_time + timedelta(days=1)).strftime('%Y-%m-%d')return datetime.strptime(f"{next_day} {self.fixed_times[0]}", "%Y-%m-%d %H:%M")async def redis_client(self):return await redis.from_url("redis://localhost:6379", decode_responses=True)async def fixed_time_task(self):while True:next_run_time = self.get_next_run_time()sleep_seconds = max(1, (next_run_time - datetime.now()).total_seconds())logger.info(f"[定时任务] 下次执行时间: {next_run_time}, sleep {sleep_seconds:.0f} 秒")await asyncio.sleep(sleep_seconds)logger.info("[定时任务] 执行具体逻辑...✅")# TODO: 添加你自己的定时任务逻辑async def check_redis_and_push(self):while True:try:redis = await self.redis_client()now_score = int(datetime.now().strftime("%H%M"))trigger_score = now_score + 4subscriptions = await redis.zrangebyscore(self.redis_subscribe_key, now_score, trigger_score, withscores=True)if subscriptions:logger.info(f"[推送检查] 检测到 {len(subscriptions)} 条订阅")for sub_key, _ in subscriptions:user_id, tag = self.parse_key(sub_key)push_key = f"{self.exist_subscribe_key}:{user_id}|{tag}"if not await redis.exists(push_key):logger.info(f"[推送中] 推送消息给用户 {user_id}，标签：{tag}")# TODO: 实际的推送逻辑await redis.setex(push_key, self.push_interval_seconds, "1")await redis.aclose()except Exception as e:logger.error(f"[推送异常] {str(e)}")await asyncio.sleep(60)def parse_key(self, key):"""解析订阅键 user:xxx|tag:xxx"""try:parts = key.split("|")user_id = parts[0].split(":")[1]tag = parts[1].split(":")[1]return user_id, tagexcept Exception as e:logger.error(f"解析 key 失败: {key} -> {e}")return "", ""if __name__ == "__main__":async def main():service = ScheduleService()await asyncio.gather(service.fixed_time_task(),service.check_redis_and_push(),)asyncio.run(main())| 方案 | 是否异步 | 优点 | 缺点 | 推荐场景 |
|---|---|---|---|---|
| 上述自定义写法 | ✅ | 灵活，Redis 任务配合好 | 需手动维护时间逻辑 | 任务量少 + redis调度系统 |
| schedule | ❌ | 简单、易用 | 不支持异步、不适合服务部署 | 脚本类、一次性任务 |
| APScheduler | ✅ | 支持异步 + cron/interval，易维护 | 初学者需要学习下语法 | 推荐服务常驻型场景 |
支持异步”，就是指在任务调度器中，能直接运行这种：
async def job():# 这里可以有 await，例如操作数据库、访问 Redis、发 HTTP 请求等await some_async_operation()print("异步任务完成")
而 APScheduler 的 AsyncIOScheduler 会让你 不需要 while True，你只要注册一次 job 函数，它会自动在对应时间点调度、支持并发、支持异步、支持 cron 等复杂逻辑，例如：
from apscheduler.schedulers.asyncio import AsyncIOSchedulerasync def job():await do_something_async()scheduler = AsyncIOScheduler()
scheduler.add_job(job, 'interval', seconds=10)
scheduler.start()
这个 wrapper() 是普通函数，apscheduler 就可以调度它，而它内部通过 asyncio.create_task() 启动了异步任务。