如何只运行 Celery schedule 多次，直到再次调用任务？答案

【问题标题】：How to run Celery schedule only a number of times and quite until the task is called again?如何只运行 Celery schedule 多次，直到再次调用任务？
【发布时间】：2019-10-09 16:20:56
【问题描述】：

我正在使用 django + celery 任务调度程序来运行每月安排一次的任务。但我只想让这个任务运行几个月，例如 3 个月或 6 个月或 9 个月..

如何阻止工作人员执行进一步的任务，然后在再次调用该任务时重新启动？

这是我的任务

@task(name="add_profit")
def count():
    portfolios = Portfolio.objects.filter(status='ACTIVE')
    if portfolios.exists():
        for portfolio in portfolios:
            user = portfolio.user
            #calculates portfolio profit
            amount = portfolio.amount * 0.1
            if portfolio.duration == '3 Months':
                PortfolioProfit.objects.create(user=user, amount=amount)
                user.useraccount.account_balance += amount
                user.useraccount.save()

这是我的芹菜任务时间表

app.conf.beat_schedule = {
    # Executes 1st day of every Month.
    'every-minute': {
        'task': 'add_profit',
        # crontab can be changes to change Schedule
        # http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html
        'schedule': crontab(0, 0, day_of_month = 1),
    },
}

【问题讨论】：

为了便于阅读，您可以将 'every-minute' 重命名为 'every-month'
我以为我理解了这个问题。由于您添加了代码 sn-ps 我不太确定。现在看来，您希望每个月都调用 celery 任务，并且某些投资组合的任务中的某些操作应该只在有限的时间内执行。到目前为止，我给你的答案是仅在一定个月内调用 celery 任务。请你澄清一下。如果您只想在有限的时间内对一个特定的投资组合执行任务，那么最好更改模型，以便您可以存储它是否仍需要调用。
基本上这些任务会被用户随机触发。根据他们选择的投资组合的持续时间，然后此任务将每月为他们的帐户余额增加一定的金额，有些是 3 个月，有些是 6 个月..等等。并且这些任务应该针对所有处于活动状态的投资组合运行。

标签： python django redis celery

【解决方案1】：

提出新的答案。我想保留前一个作为记录，因为它可能会帮助其他遇到不同问题的人。对于您的问题，我认为您只需每月无条件地调用 celery 任务即可。

我认为最简单的方法是更改存储在数据库中的数据，以便您可以识别活跃的投资组合，存储第一次的日期和最后一次自动添加值的日期。

现在，每月 celery 任务将识别 first_date

【讨论】：

谢谢。我认为这种模式应该有效，但是如果我在最后一个日期使用 timedelta 值，它会有效吗？
在这种情况下也可以正常工作：first_date <= today <= first_date + time_delta

【解决方案2】：

选项 1：您每个月都使用 celery crontab 条目开始任务，并在任务中添加测试：如果当前日期不在特定范围内，您只需退出处理。

这有一点开销，但每月一次的视线开销应该是可以接受的。

@task(name="add_profit")
def count():
    today = datetime.datetime.now()
    if today > datetime.datetime(2020,1, 1):
        return
    # the remaining part of your task follows here

选项 2：您执行一次（使用或不使用 for 循环）一个小代码 sn-p，用于安排相关月份的任务。

http://docs.celeryproject.org/en/latest/userguide/calling.html#eta-and-countdown

在下面的示例中，我只是展示了为接下来的三天安排任务的想法：

today = datetime.utcnow()

for delta in range(1, 4):
    task.apply_async(args=[arg1, arg2, ...), eta=today + timedelta(days=delta))

在这里，任务将按照您想要的频率执行。

但是，如果您将服务器迁移到其他位置，如果您重置 rabbitmq（或您拥有的任何代理），则要安排的任务将丢失/消失

【讨论】：

嗨@gelonida，感谢您的回复。
我刚刚更新了我的问题并添加了代码，因为我只是一个初学者，所以我很不确定如何实施您的解决方案。我显示了任务和调度程序。如果portfolio.duration 为3 个月，则从任务中，我想运行此任务3 次并退出计划。我如何做到这一点？

【解决方案3】：

好的。非常感谢@gelonida 您的意见。我已经能够实现我的目标。

我创建了到期日期的数据库输入（3 个月、6 个月等，取决于用户的选择），它是使用投资组合的创建日期和未来某个时间计算的，如下所示：

portfolio.expiry_date = timezone.now() + timedelta(days = 93) # for 3months

然后在我的任务中我这样做了

from django.utils import timezone


@task(name="add_profit")
def count():
    portfolios = Portfolio.objects.filter(status='ACTIVE')
    current_datetime = timezone.now()
    if portfolios.exists():
        for portfolio in portfolios:
            if current_datetime > portfolio.expiry_date:
                portfolio.status = 'COMPLETED'
                portfolio.save()
                return
            else:
                user = portfolio.user
                amount = portfolio.amount * 0.1
                PortfolioProfit.objects.create(user=user, amount=amount)
                user.useraccount.account_balance += amount
                user.useraccount.save()

这对我来说非常有效。

【讨论】：