如何使用Celery调度每月最后一天执行的任务（适配不同天数月份）

嘿，这个问题确实有点棘手，毕竟标准crontab表达式没法直接适配所有月份的最后一天——毕竟有的月只有28天，有的却有31天。不过别担心，我有两个靠谱的方案帮你搞定：

方案一：用Celery内置的crontab扩展值L #

Celery对标准crontab做了扩展，支持用L来表示「每月最后一天」，它会自动识别每个月的天数，完美适配28/29/30/31天的情况。

直接在beat_schedule里这样配置就行：

from celery import Celery
from celery.schedules import crontab

app = Celery('tasks', broker='pyamqp://guest@localhost//')

# 配置定时任务
app.conf.beat_schedule = {
    'run-on-monthly-last-day': {
        'task': 'tasks.your_target_task',  # 替换成你的任务路径
        'schedule': crontab(hour=2, minute=0, day_of_month='L'),  # 每月最后一天凌晨2点执行
        'args': (),  # 任务需要的参数，没有就留空
    },
}

# 别忘了设置时区，避免时间偏移
app.conf.timezone = 'Asia/Shanghai'  # 替换成你的业务时区

这个方案最简单，只要你的Celery版本不是特别老（基本v4.0以上都支持），直接用就行。

方案二：自定义调度类（适合更灵活的场景） #

如果你需要更复杂的逻辑（比如加上额外的执行条件，或者处理特殊时区场景），可以自己写一个调度类，继承Celery的Schedule类来实现逻辑。

比如下面这个例子，会自动计算每个月的最后一天，并判断当前是否到了执行时间：

from datetime import datetime, timedelta
from celery import Celery
from celery.schedules import Schedule

app = Celery('tasks', broker='pyamqp://guest@localhost//')

class MonthlyLastDaySchedule(Schedule):
    def __init__(self, hour=0, minute=0):
        self.hour = hour  # 执行小时
        self.minute = minute  # 执行分钟

    def is_due(self, last_run_at):
        now = datetime.now()
        # 计算当前月的最后一天：先跳到下个月，再减去当月天数
        next_month = now.replace(day=28) + timedelta(days=4)  # 确保进入下个月
        current_last_day = next_month - timedelta(days=next_month.day)
        # 设置执行时间为指定的时分
        next_run_time = current_last_day.replace(
            hour=self.hour, minute=self.minute, second=0, microsecond=0
        )

        # 判断是否需要立即执行
        if now.date() == current_last_day.date():
            if now.time() < next_run_time.time():
                # 今天就是最后一天，且还没到执行时间，马上执行
                delay = (next_run_time - now).total_seconds()
                return (True, delay)
            else:
                # 今天已经过了执行时间，等下个月的最后一天
                next_month = current_last_day.replace(day=28) + timedelta(days=4)
                next_last_day = next_month - timedelta(days=next_month.day)
                next_run_time = next_last_day.replace(
                    hour=self.hour, minute=self.minute, second=0, microsecond=0
                )
                delay = (next_run_time - now).total_seconds()
                return (False, delay)
        else:
            # 还没到当月最后一天，计算等待时间
            delay = (next_run_time - now).total_seconds()
            return (False, delay)

# 注册自定义调度的任务
app.conf.beat_schedule = {
    'custom-monthly-last-day-task': {
        'task': 'tasks.your_target_task',
        'schedule': MonthlyLastDaySchedule(hour=2, minute=0),
        'args': (),
    },
}

app.conf.timezone = 'Asia/Shanghai'

这个方案的优势是完全可控，你可以根据自己的需求修改逻辑，比如加上节假日判断、时区转换等。

小提醒 #

不管用哪个方案，都记得设置正确的时区，否则任务执行时间可能会和你预期的有偏差哦。

内容的提问来源于stack exchange，提问作者Sonam Agrawal