Правильная настройка django redis celery и celery beats

У меня были эти проблемы раньше. Это не твой код. Обычно это проблема с окружающей средой. Вы должны запустить все в virtualenv, добавив requirements.txt файл с конкретными версиями пакетов.

Существует известная проблема, касающаяся celery 4.x и django 1.x, поэтому вам следует подумать о пакетах, которые вы используете.

В этом руководстве подробно объясняется, как построить virtualenv из сельдерея.

Если вы можете сказать мне версии своих пакетов, я могу попробовать помочь другим способом.

Изменить:

Думаю, дело в том, как вы управляете сельдереем. Если мы исправили первую проблему, попробуйте поиграть с этим:

celery -A aenima.celery:app beat -l debug --scheduler django_celery_beat.schedulers:DatabaseScheduler

or

celery -A aenima.aenima.celery:app beat -l debug --scheduler django_celery_beat.schedulers:DatabaseScheduler

Последняя ошибка, которую вы получаете, связана с обнаружением вашего модуля. Сначала попробуйте.

Было бы удобно использовать virtualenv.

Во-первых, как сказал @Gal, вам нужно убедиться, что у вас есть celery 4.x.

Вы можете установить это, выполнив pip:

pip install сельдерей

Конечно, вы также можете установить 4.x версию, добавив ее в свой requirements.txt следующим образом:

сельдерей == 4.1.0

Или более поздние версии, если они будут доступны в будущем.

Затем вы можете переустановить все свои пакеты, используя:

• pip install -r requirements.txt

Это гарантирует, что у вас установлен этот определенный пакет сельдерея.

Теперь о Celery, хотя ваш код может и не ошибочен, но я напишу так, как я заставил мое приложение Celery работать.

__init __.py:

from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery_conf import app as celery_app

__all__ = ['celery_app']

celery_conf.py:

from __future__ import absolute_import, unicode_literals

import os

from celery import Celery
from datetime import timedelta

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '<PATH.TO.YOUR.SETTINGS>')

app = Celery('tasks')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

# Set a beat schedule to update every hour.
app.conf.beat_schedule = {
    'update-every-hour': {
        'task': 'tasks.update',
        'schedule': timedelta(minutes=60),
        'args': (16, 16),
    },
}

# The default task that Celery runs.
@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

tasks.py:

# -*- coding: utf-8 -*-
from __future__ import unicode_literals

import requests

from django.conf import settings
from django.http import HttpResponse

from celery.task import Task
from celery.five import python_2_unicode_compatible
from celery import Celery
app = Celery()


@python_2_unicode_compatible
class Update(Task):
    name = 'tasks.update'

    def run(self, *args, **kwargs):
        # Run the task you want to do.

""" For me the regular TaskRegistry didn't work to register classes, 
so I found this handy TaskRegistry demo and made use of it to 
register tasks as classes."""
class TaskRegistry(Task):

    def NotRegistered_str(self):
        self.assertTrue(repr(TaskRegistry.NotRegistered('tasks.add')))

    def assertRegisterUnregisterCls(self, r, task):
        with self.assertRaises(r.NotRegistered):
            r.unregister(task)
        r.register(task)
        self.assertIn(task.name, r)

    def assertRegisterUnregisterFunc(self, r, task, task_name):
        with self.assertRaises(r.NotRegistered):
            r.unregister(task_name)
        r.register(task, task_name)
        self.assertIn(task_name, r)

    def task_registry(self):
        r = TaskRegistry()
        self.assertIsInstance(r, dict, 'TaskRegistry is mapping')

        self.assertRegisterUnregisterCls(r, Update)

        r.register(Update)
        r.unregister(Update.name)
        self.assertNotIn(Update, r)
        r.register(Update)

        tasks = dict(r)
        self.assertIsInstance(
            tasks.get(Update.name), Update)

        self.assertIsInstance(
            r[Update.name], Update)

        r.unregister(Update)
        self.assertNotIn(Update.name, r)

        self.assertTrue(Update().run())

    def compat(self):
        r = TaskRegistry()
        r.regular()
        r.periodic()

Как я объяснил в коде, обычный taskregistry не работал, встроенный в Celery 4.x, поэтому я использовал демонстрационный реестр задач. Конечно, вы также можете не использовать классы для выполнения задач, но я предпочел использовать классы.

settings.py:

# Broker settings for redis
CELERY_BROKER_HOST = '<YOUR_HOST>'
CELERY_BROKER_PORT = 6379
CELERY_BROKER_URL = 'redis://'
CELERY_DEFAULT_QUEUE = 'default'

# Celery routes
CELERY_IMPORTS = (
    'PATH.TO.tasks' # The path to your tasks.py
)

CELERY_DATABASE_URL = {
    'default': '<CELERY_DATABASE>', # You can also use your already being used database here
}

INSTALLED_APPS = [
    ...
    'PATH.TO.TASKS' # But exclude the tasks.py from this path
]

LOGGING = {
    ...
    'loggers': {
        'celery': {
            'level': 'DEBUG',
            'handlers': ['console'],
            'propagate': True,
        },
    }
}

Я запускаю свой рабочий с помощью следующих команд:

redis-server --daemonize да

celery multi start worker -A PATH.TO.TASKS -l info --beat # Но исключить tasks.py из пути

Я надеюсь, что эта информация может помочь вам или всем, кто борется с сельдереем.

РЕДАКТИРОВАТЬ:

Обратите внимание, что я запускаю рабочий как демон, поэтому вы не сможете увидеть журналы в консоли. Для меня это записано в .txt файл.

Также обратите внимание на пути, которые можно использовать, например, для некоторых, вам нужно включить путь к вашему приложению, например:

project.apps.app

А для других случаев вам нужно включить tasks.py без .py, я написал, когда исключать этот файл, а когда нет.

РЕДАКТИРОВАТЬ 2:

Декоратор @shared_task возвращает прокси, который всегда использует экземпляр задачи в current_app. Это делает декоратор @shared_task полезным для библиотек и приложений многократного использования, поскольку у них не будет доступа к приложению пользователя.

Обратите внимание, что @shared_task не имеет доступа к приложению пользователя. Приложение, которое вы сейчас пытаетесь зарегистрировать, не имеет доступа к вашему приложению. Метод, который вы действительно хотите использовать для регистрации задачи:

from celery import Celery
app = Celery()

@app.task
def test_celery(x, y):
    logger = logging.getLogger("AENIMA")
    print("EUREKA")
    logger.debug("EUREKA")

Получено незарегистрированное задание типа u'tasks.test_celery '. Сообщение было проигнорировано и отклонено.

Вы не забыли импортировать модуль, содержащий эту задачу? Или, может быть, вы используете относительный импорт?

Возможно, ваш путь к задаче неверен, должен быть:

app.conf.beat_schedule = {
    'test-every-30-seconds': {
        'task': 'criptoball.tasks.test_celery',
        'schedule': 30.0,
        'args': (16, 16)
    }, 
}

tasks.test_celery должен быть полным путем ： criptoball.tasks.test_celery

Вам следует исправить одну вещь: используйте:

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

чтобы сообщить Celery, какие задачи приложений вы хотите, чтобы он обнаруживал, если вы используете Celery 3.x.