在存储在数据库中的时间运行周期性任务答案

【问题标题】：Running periodic task at time stored in database在存储在数据库中的时间运行周期性任务
【发布时间】：2018-10-22 02:21:15
【问题描述】：

目前，我在 Azure 实例上使用作业计划程序设置了定期任务。这些是在固定时间触发 API (Django) 端点。

我想让这些时间动态化（这不适用于此解决方案）。计划是直接从 Django 触发这些任务。计划时间将存储在我的数据库 (MySQL) 中并被检索以创建计划作业。当这些值改变时，调度器也应该相应地改变。

在查看 Celery 之后，似乎使用 periodic tasks crontab schedules 可以工作。 使用它，是否可以根据我的数据库中的值设置我的预定时间？

看来我还需要一个 Redis 实例。 因为我只会将 Celery 用于定期任务，这仍然是正确的方法吗？

【问题讨论】：

标签： python django celery django-celery celerybeat

【解决方案1】：

Cron 被认为是运行定期发生的任务，因为它具有简单的配置选项，每天、每小时、每 15 分钟...
添加 cron 作业并不是将动态任务配置为在具体日期（或日期时间）运行的好方法

您可以按照this 答案中的说明使用schedule 模块。

还有另一个库apscheduler，但是检查最后一个版本是否适用于python3（如果你使用它）

from datetime import date
from apscheduler.scheduler import Scheduler

# Start the scheduler
sched = Scheduler()
sched.start()

# Define the function that is to be executed
def my_job(text):
    print text

# The job will be executed on November 6th, 2009
exec_date = date(2009, 11, 6)

# Store the job in a variable in case we want to cancel it
job = sched.add_date_job(my_job, exec_date, ['hello'])

【讨论】：

【解决方案2】：

我正在使用Django、Celery、RabbitMQ 和 postgreSQL。

我正在做你想做的事。

画中画：celery 和 flower

您需要一个 Celery conf 文件（在您的 settings.py 文件夹中）：

你要添加的是beat_schedule：

app.conf.beat_schedule = {
    'task-name': {
        'task': 'myapp.tasks.task_name',
        'schedule': crontab(minute=30, hour=5, day_of_week='mon-fri'),
    },
}

这将在您的数据库中添加一个条目以执行task_name（周一至周五5:30），您可以直接更改您的设置（重新加载celery和celery beat之后）

我喜欢的是你可以很容易地添加重试机制，并且安全：

@app.task(bind=True, max_retries=50)
def task_name(self, entry_pk):
    entry = Entry.objects.get(pk=entry_pk)
    try:
        entry.method()
    except ValueError as e:
        raise self.retry(exc=e, countdown=5 * 60, queue="punctual_queue")

当我的method() 提升ValueError 时，我将在 5 分钟内重新执行此方法，最多尝试 50 次。

好处是你可以在 Django admin 中访问数据库：

您可以通过flower检查任务是否执行（带有回溯）：

我每天要执行超过 1000 个任务，你需要的是创建队列和工作者。

我为此使用了 10 名工人（为了将来扩大规模）：

celery multi start 10 -A MYAPP -Q:1-3 recurring_queue,punctual_queue -Q:4,5 punctual_queue -Q recurring_queue --pidfile="%n.pid"

以及启动任务的守护进程：

celery -A MYAPP beat -S django --detach

这对你来说可能有点过头了，但他可以为你做更多的事情： - 异步发送电子邮件（如果失败，您可以更正并重新发送电子邮件） - 为用户上传和后处理异步 - 每一项需要时间但您不想等待的任务（您可以将需要完成的任务链接起来返回一个结果并在另一个任务中使用它）

【讨论】：

感谢您的详细回答。需要发生的一件关键事情是能够更改 chrontab 中的参数以从数据库中提取（动态值）？我想将分钟、小时和 day_of_week 值设置为从数据库中提取的值。你知道这个解决方案是否可行？
Beat schedule (celery Beat) 在你启动守护进程时写入数据库，但 cron 将使用存储在数据库中的值启动（你可以动态更新）

【解决方案3】：

在没有外部库的情况下，设置一个每日 cron 脚本，从数据库中获取今天的任务，并在当时使用线程来运行。

def take_a_background_nap(time_to_send):
    while datetime.datetime.now() < time_to_send:
        time.sleep(60)
    print('finally running')
    return


threadObj = threading.Thread(target=take_a_background_nap, args=[datetime.datetime(2020, 5, 11, 12, 53, 0)],)
threadObj.start()

您可以拥有任意数量的线程，但要注意并发问题。

【讨论】：