python - 如何从以前的 celery 任务中产生 celery 任务？

Question

我可能用错了 celery 。但是我正在开发的聊天机器人需要 celery 和 redis 来完成异步任务。这是我正在使用的框架:http://microsoftbotframework.readthedocs.io/en/latest/asynctasks/ .

我的特定用例目前要求我永远 运行 celery 任务，并在两者之间等待一段任意时间，范围从 30 分钟到 3 天不等。像这样的东西

@celery.task
def myAsyncMethod():
    while true:
        timeToWait = getTimeToNextAlarm()
        sleep(timeToWait)
        sendOutMessages()

基本上，我有一个永不退出的异步进程。我很确定不应该像这样使用 celery 。 所以我的问题是，我如何创建一个 celery 任务来处理第一个任务、生成一个任务并将其提交到 celery 队列并退出。基本上是这样的:

@celery.task
def myImprovedTask():
    timeToWait = getTimeToNextAlarm()
    sleep(timeToWait)
    sendOutMessages()
    myImprovedTask().delay()    # recursive call to async method for next event

不一定是递归的或什至像这样，但这是 celery 最初打算使用的方式(我相信是为了短期任务？)

Tl;dr:如何从另一个任务中创建 celery 任务并退出原始任务？

请告诉我是否应该进一步解释。谢谢。

Answer 1

如果您想从初始任务运行另一个任务，只需像通常使用 Task.delay() 那样调用它即可, 或 Task.apply_async() :

@celery.task
def myImprovedTask():
    timeToWait = getTimeToNextAlarm()
    sleep(timeToWait)
    sendOutMessages()
    myImprovedTask.delay()

再次调用同一个任务也没关系。它使用 delay() 入队，您的原始任务返回，然后运行队列中的下一个任务。

---

所有这些都是在假设您实际上是异步调用您的 Celery 任务。有时情况并非如此，一个常见的罪魁祸首是 task-always-eager配置选项。默认情况下它是禁用的，但是(来自 docs ):

If task_always_eager is True, all tasks will be executed locally by blocking until the task returns. apply_async() and Task.delay() will return an EagerResult instance, that emulates the API and behavior of AsyncResult, except the result is already evaluated.

That is, tasks will be executed locally instead of being sent to the queue.

因此，请确保您的 Celery 配置包括:

task_always_eager = False