- r - 以节省内存的方式增长 data.frame
- ruby-on-rails - ruby/ruby on rails 内存泄漏检测
- android - 无法解析导入android.support.v7.app
- UNIX 域套接字与共享内存(映射文件)
我在使用 current_app.send_task 时遇到 Celery 队列路由问题
我有两个 worker (每个队列一个)
python manage.py celery worker -E -Q priority --concurrency=8 --loglevel=DEBUG
python manage.py celery worker -Q low --concurrency=8 -E -B --loglevel=DEBUG
我在 celeryconfig.py 文件中定义了两个队列:
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.core.exceptions import ImproperlyConfigured
from celery import Celery
from django.conf import settings
try:
app = Celery('proj', broker=getattr(settings, 'BROKER_URL', 'redis://'))
except ImproperlyConfigured:
app = Celery('proj', broker='redis://')
app.conf.update(
CELERY_TASK_SERIALIZER='json',
CELERY_ACCEPT_CONTENT=['json'],
CELERY_RESULT_SERIALIZER='json',
CELERY_RESULT_BACKEND='djcelery.backends.database:DatabaseBackend',
CELERY_DEFAULT_EXCHANGE='tasks',
CELERY_DEFAULT_EXCHANGE_TYPE='topic',
CELERY_DEFAULT_ROUTING_KEY='task.priority',
CELERY_QUEUES=(
Queue('priority',routing_key='priority.#'),
Queue('low', routing_key='low.#'),
),
CELERY_DEFAULT_EXCHANGE='priority',
CELERY_IMPORTS=('mymodule.tasks',)
CELERY_ENABLE_UTC = True
CELERY_TIMEZONE = 'UTC'
if __name__ == '__main__':
app.start()
在任务的定义中,我们使用装饰器显式队列:
@task(name='mymodule.mytask', routing_key='low.mytask', queue='low')
def mytask():
# does something
pass
当此任务使用以下方式运行时,此任务确实在低队列中运行:
from mymodule.tasks import mytask
mytask.delay()
但是当它运行时不是这样的:(它在默认队列中运行:“优先级”)
from celery import current_app
current_app.send_task('mymodule.mytask')
我想知道为什么后面的方法不将任务路由到“低”队列!
p.s: 我用的是redis
最佳答案
send_task 是一个低级方法。它直接向代理发送任务签名,而无需通过您的任务装饰器。使用此方法,您甚至可以在不加载任务代码/模块的情况下发送任务。
要解决您的问题,您可以直接从配置中获取 routing_key/queue:
route = celery.amqp.routes[0].route_for_task("mymodule.mytask")
Out[10]: {'queue': 'low', 'routing_key': 'low.mytask'}
celery.send_task("myodule.mytask", queue=route['queue'], routing_key=route['routing_key']`
关于python - 使用 current_app.send_task 时,Celery 任务路由不起作用,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/27965672/
今天有小伙伴给我留言问到,try{...}catch(){...}是什么意思?它用来干什么? 简单的说 他们是用来捕获异常的 下面我们通过一个例子来详细讲解下
我正在努力提高网站的可访问性,但我不知道如何在页脚中标记社交媒体链接列表。这些链接指向我在 facecook、twitter 等上的帐户。我不想用 role="navigation" 标记这些链接,因
说现在是 6 点,我有一个 Timer 并在 10 点安排了一个 TimerTask。之后,System DateTime 被其他服务(例如 ntp)调整为 9 点钟。我仍然希望我的 TimerTas
就目前而言,这个问题不适合我们的问答形式。我们希望答案得到事实、引用资料或专业知识的支持,但这个问题可能会引发辩论、争论、投票或扩展讨论。如果您觉得这个问题可以改进并可能重新打开,visit the
我就废话不多说了,大家还是直接看代码吧~ ? 1
Maven系列1 1.什么是Maven? Maven是一个项目管理工具,它包含了一个对象模型。一组标准集合,一个依赖管理系统。和用来运行定义在生命周期阶段中插件目标和逻辑。 核心功能 Mav
我是一名优秀的程序员,十分优秀!