时间戳

首页 > 解决方案 > 路由 celery 任务

python - 路由 celery 任务

问题描述

尝试创建两个单独的专用工作人员时，我无法向 celery 发送任务。我已经阅读了文档和这个问题，但它并没有改善我的情况。

我的配置如下：

CELERY_RESULT_BACKEND = 'django-db'
CELERY_BROKER_URL = f'redis://{env("REDIS_HOST")}:{env("REDIS_PORT")}/{env("REDIS_CELERY_DB")}'
CELERY_DEFAULT_QUEUE = 'default'
CELERY_DEFAULT_EXCHANGE_TYPE = 'topic'
CELERY_DEFAULT_ROUTING_KEY = 'default'
CELERY_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('media', Exchange('media'), routing_key='media'),
)
CELERY_ROUTES = {
    'books.tasks.resize_book_photo': {
        'queue': 'media',
        'routing_key': 'media',
    },
}

任务在tasks.py文件中定义如下：

import logging
import time

from celery import shared_task


from books.models import Author, Book
from books.commands import resize_book_photo as resize_book_photo_command


logger = logging.getLogger(__name__)


@shared_task
def list_test_books_per_author():
    time.sleep(5)
    queryset = Author.objects.all()
    for author in queryset:
        for book in author.testing_books:
            logger.info(book.title)


@shared_task
def resize_book_photo(book_id: int):
    resize_book_photo_command(Book.objects.get(id=book_id))

他们被称为使用apply_async：

list_test_books_per_author.apply_async()
resize_book_photo.apply_async((book.id,))

当我运行 celery flower 时，我看到队列中没有出现任何任务。 芹菜花面板显示两个没有任务的活跃工人。

工人开始使用：

celery -A blacksheep worker -l info --autoscale=10,1 -Q media --host=media@%h
celery -A blacksheep worker -l info --autoscale=10,1 -Q default --host=default@%h

我可以做的是使用redis-cli和127.0.0.1:6379> LRANGE celery 1 100命令确认它们最终在celerykey 下（这是 celery 的默认值）。似乎没有工人消费。

编辑在仔细查看这部分文档后，我注意到我的命名是错误的。将设置更改为：

CELERY_RESULT_BACKEND = 'django-db'
CELERY_BROKER_URL = f'redis://{env("REDIS_HOST")}:{env("REDIS_PORT")}/{env("REDIS_CELERY_DB")}'
CELERY_TASK_DEFAULT_QUEUE = 'default'
# CELERY_DEFAULT_EXCHANGE_TYPE = 'topic'
CELERY_TASK_DEFAULT_ROUTING_KEY = 'default'
CELERY_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('media', Exchange('media'), routing_key='media'),
)
CELERY_ROUTES = {
    'books.tasks.resize_book_photo': {
        'queue': 'media',
        'routing_key': 'media',
    },
}

情况有所改善：任务从队列中消耗default，但我想去队列的任务media也去default.

EDIT2我试图通过将某个任务的调用更改为resize_book_photo.apply_async((book.id,), queue='media'). 该任务已正确分派到正确的队列并被使用。但是，我更喜欢自动处理这个问题，这样我就不必在调用时定义队列apply_async

标签： pythondjangocelery

解决方案

尝试CELERY_TASK_ROUTES代替CELERY_ROUTES. 这最近对我有用 django 集成。

解释隐藏在此评论中：How to route tasks to different queues with Celery and Django

推荐阅读