时间戳

首页 > 解决方案 > 如何运行一组 celery 任务？应用尚未加载

python - 如何运行一组 celery 任务？应用尚未加载

问题描述

例如，我有两个任务：a和b. 我需要并行运行它们。我创建了一组任务并尝试运行它。但我得到错误

项目/应用程序/tasks.py

@app.task
def a():
    pass

@app.task
def b():
    pass

django.core.exceptions.AppRegistryNotReady：应用程序尚未加载。

申请app已注册INSTALLED_APPS并完成所有迁移

项目/项目/__init__.py

from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app

__all__ = ('celery_app',)

项目/项目/芹菜.py

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

app = Celery('proj')
app.config_from_object('django.conf:settings', namespace='CELERY')

app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))


from celery import group
from app.tasks import a, b

run_group = group(a.s(), b.s())
run_group()

追溯

File "/home/m0nte-cr1st0/test_projects/proj/proj/__init__.py", line 5, in <module>
    from .celery import app as celery_app
File "/home/m0nte-cr1st0/test_projects/proj/proj/celery.py", line 26, in <module>
    from app.tasks import a, b
File "/home/m0nte-cr1st0/test_projects/proj/app/tasks.py", line 14, in <module>
    from .models import Link, Prediction, PredictionBK
File "/home/m0nte-cr1st0/test_projects/proj/app/models.py", line 2, in <module>
    from django.contrib.auth.models import AbstractUser

标签： pythondjangocelery

解决方案

项目/项目/芹菜.py

import os
from celery import Celery
from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

app = Celery('proj')
app.config_from_object('django.conf:settings')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

项目/项目/settings.py

REDIS_HOST = 'localhost'
REDIS_PORT = '6379'
BROKER_URL = 'redis://' + REDIS_HOST + ':' + REDIS_PORT + '/0'
BROKER_TRANSPORT_OPTIONS = {'visibility_timeout': 3600}
CELERY_RESULT_BACKEND = 'redis://' + REDIS_HOST + ':' + REDIS_PORT + '/0'

项目/项目/__init__.py

from .celery import app as celery_app

项目/应用程序/tasks.py

from celery import group


@app.task
def a():
    pass


@app.task
def b():
    pass


run_group = group(a.s(), b.s())
run_group()

推荐阅读