基于Django的Web应用：如何在指定时间段运行Celery任务？

嘿，这个需求我之前也碰到过，结合Django+Celery的场景，给你几个实用的实现思路：

1. 直接用Celery Beat的Crontab表达式精准触发 #

如果你的任务需要在每天14:00-15:00之间定时执行（比如每分钟跑一次），最直接的方式就是通过Celery Beat的crontab配置来指定时间范围，不需要额外的判断逻辑。

举个配置例子，在你的Celery配置文件（比如celery.py）里添加：

from celery.schedules import crontab

# 配置Beat定时任务
app.conf.beat_schedule = {
    # 每天14点整到14点59分，每分钟触发一次任务
    'task-run-between-14-15': {
        'task': 'your_app.tasks.your_target_task',  # 替换成你的任务路径
        'schedule': crontab(hour='14', minute='*'),
        # 可选：如果需要指定特定日期，还可以加day_of_month、day_of_week参数
        # 'args': (your_args,)  # 如果任务需要传参的话
    },
}

这种方式的好处是完全依赖Celery的定时调度，不需要修改任务本身，适合周期性的短任务。

2. 任务内部添加时间段校验 #

如果你的任务触发频率更高（比如每分钟都跑），但只想在14:00-15:00这个窗口内执行核心逻辑，那可以在任务开头加时间判断，不在时间段内直接跳过。

示例代码：

from datetime import datetime, time
from celery import shared_task

@shared_task
def your_target_task():
    # 定义允许执行的时间段
    allowed_start = time(14, 0)
    allowed_end = time(15, 0)
    current_time = datetime.now().time()

    # 检查当前时间是否在允许范围内
    if not (allowed_start <= current_time <= allowed_end):
        print("当前不在指定时间段，跳过任务执行")
        return

    # 这里写你的核心任务逻辑
    print("在指定时间段内，执行任务...")
    # do something...

这种方式灵活性更高，哪怕Beat的调度规则调整，只要时间段不变，任务就只会在窗口内干活。

3. 持续运行任务的场景（从14点到15点一直跑） #

如果你的任务需要从14点开始持续运行，到15点自动停止，可以让Beat在14点整触发任务，然后任务内部做循环，每次循环检查当前时间，超出范围就退出。

示例代码：

from datetime import datetime, time
import time as t
from celery import shared_task

@shared_task
def run_task_during_window():
    start_time = time(14, 0)
    end_time = time(15, 0)

    while True:
        current_time = datetime.now().time()
        if current_time > end_time:
            print("已超出指定时间段，停止任务")
            break

        # 执行单次任务逻辑
        print("正在执行任务...")
        # do something...

        # 间隔一段时间再执行下一次（比如1分钟）
        t.sleep(60)

注意：这种长时间运行的任务，需要确保Celery Worker的超时设置（比如--time-limit）大于1小时，避免Worker把任务强制杀掉。

根据你的实际需求选对应的方案就好啦！

内容的提问来源于stack exchange，提问作者Sandeep Prasad Kushwaha