定时任务(python)

2025-03-02 | 阅读：次

介绍

🧩 什么是“定时任务”？

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

•	每天早上8点发通知
•	每隔10秒采集一次数据
•	每小时清理一次缓存

相关使用

✅ 最简单的方式：while True + time.sleep()

import time

def job():
    print("执行任务")

while True:
    job()
    time.sleep(10)  # 每10秒执行一次

✅ 优点：

• 写法简单，不需要任何依赖

• 控制力强

❌ 缺点：

• 会阻塞当前线程

• 精度差（任务执行时间会影响间隔）

• 没有“精确到几点几分”的调度能力

✅ 进阶方案：使用 schedule 库

pip install schedule

import schedule
import time

def 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 BlockingScheduler

def 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, timedelta

import redis.asyncio as redis

logging.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 + 4

                subscriptions = 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, tag
        except 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 AsyncIOScheduler

async def job():
    await do_something_async()

scheduler = AsyncIOScheduler()
scheduler.add_job(job, 'interval', seconds=10)
scheduler.start()