1

我正在尝试编写一个执行 celery 任务的自定义 django-admin 命令，但是该任务没有执行，并且当我尝试时 django 只是挂起。

from django.core.management.base import BaseCommand
from myapp.tasks import my_celery_task

class Command(BaseCommand):
    def handle(self, *args, **options):
        print "starting task"
        my_celery_task.delay()
        print "task has been sent"

调用命令时我收到的输出是：

starting task

我从来没有到达“任务已发送”这一行。它只是挂起。我不确定为什么任务没有运行。当被视图调用时，芹菜任务被完美地调用。

4

2 回答 2

0

这似乎适用于 celery 5.1.2，但如果它是shared_task

======任务.py

from celery import current_app
from celery.schedules import crontab
app = current_app._get_current_object()

@app.task
def my_celery_task():
    # Do something interesting
    return 0 # Return the result of this operation

====== my_command.py

from accounts.tasks import my_celery_task
...
    def handle(self, *args, **options):
        self.stdout.write("Queing request")
        result = my_celery_task.delay()
        self.stdout.write("Waiting for task to end")
        updated = result.wait(timeout=120.0)

        self.stdout.write(f"Returned {updated}")

而且，如果我将任务放在celery.py文件中，我最终会遇到死锁，这是一个要调试的 PITA

于 2022-02-05T23:24:14.980 回答
-1

升级到 High Sierra 后，问题实际上出在 Mac 上的 RabbitMQ 上。

于 2017-12-05T15:02:40.943 回答