关于设计图的网站网站新闻百度收录

介绍

🧩 什么是“定时任务”？

定时任务，就是按照设定的时间间隔或时间点自动执行某些操作。比如：

•	每天早上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() 启动了异步任务。