У меня есть два проекта Django, каждый с приложением Celery:
- fooproj.celery_app
- barproj.celery_app
Каждое приложение запускает свой собственный работник Celery:
celery worker -A fooproj.celery_app -l info -E -Q foo_queue
celery worker -A barproj.celery_app -l info -E -Q bar_queue
Вот как я настраиваю свои приложения Celery:
import os
from celery import Celery
from django.conf import settings
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings.local')
app = Celery('celery_app', broker=settings.BROKER_URL)
app.conf.update(
CELERY_ACCEPT_CONTENT=['json'],
CELERY_TASK_SERIALIZER='json',
CELERY_RESULT_SERIALIZER='json',
CELERY_RESULT_BACKEND='djcelery.backends.database:DatabaseBackend',
CELERY_SEND_EVENTS=True,
CELERY_DEFAULT_QUEUE=settings.CELERY_DEFAULT_QUEUE,
CELERY_DEFAULT_EXCHANGE=settings.CELERY_DEFAULT_EXCHANGE,
CELERY_DEFAULT_ROUTING_KEY=settings.CELERY_DEFAULT_ROUTING_KEY,
CELERY_DEFAULT_EXCHANGE_TYPE='direct',
CELERY_ROUTES = ('proj.celeryrouters.MainRouter', ),
CELERY_IMPORTS=(
'apps.qux.tasks',
'apps.lorem.tasks',
'apps.ipsum.tasks',
'apps.sit.tasks'
),
)
Мой класс роутера:
from django.conf import settings
class MainRouter(object):
"""
Routes Celery tasks to a proper exchange and queue
"""
def route_for_task(self, task, args=None, kwargs=None):
return {
'exchange': settings.CELERY_DEFAULT_EXCHANGE,
'exchange_type': 'direct',
'queue': settings.CELERY_DEFAULT_QUEUE,
'routing_key': settings.CELERY_DEFAULT_ROUTING_KEY,
}
fooproj имеет настройки:
BROKER_URL = redis://localhost:6379/0
CELERY_DEFAULT_EXCHANGE = 'foo_exchange'
CELERY_DEFAULT_QUEUE = 'foo_queue'
CELERY_DEFAULT_ROUTING_KEY = 'foo_routing_key'
barproj имеет настройки:
BROKER_URL = redis://localhost:6379/1
CELERY_DEFAULT_EXCHANGE = 'foo_exchange'
CELERY_DEFAULT_QUEUE = 'foo_queue'
CELERY_DEFAULT_ROUTING_KEY = 'foo_routing_key'
Как видите, оба проекта используют собственную базу данных Redis в качестве брокера, свою собственную базу данных MySQL как бэкэнд, свой собственный обмен, очередь и ключ маршрутизации.
Я пытаюсь запустить два процесса событий Celery, по одному для каждого приложения:
celery events -A fooproj.celery_app -l info -c djcelery.snapshot.Camera
celery events -A barproj.celery_app -l info -c djcelery.snapshot.Camera
Проблема в том, что оба процесса события сельдерея собирают задачи от всех моих рабочих сельдерея! Итак, в базе данных fooproj я могу видеть результаты задачи из базы данных barproj.
Есть идеи, как решить эту проблему?