【发布时间】：2014-08-31 02:23:58
【问题描述】：
我有两种工人：
- 第一种链接到 django 进行数据库轮询
- 第二种不与 django 交互。
我使用rabbitMQ 作为代理，并在同一台服务器上首次启动这两种工作人员。
我想将类型 1 和类型 2 之间的代码分开，以便将来能够在单独的服务器中启动工作人员。我想避免在运行第二种工作者的服务器中使用 django 代码。
在 django worker 方面，我这样定义它们：
@shared_task(name='task_polling')
def task_polling():
send_task('do_work', [], {'_serialized_task': serialized_task})
@shared_task(name='save_shell_task')
def save_shell_task(_result):
pass
还有如下 celery 配置：
from __future__ import absolute_import
import os
from datetime import timedelta
from celery import Celery
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'ussd_auto.settings.local')
app = Celery(include=[
'app_task_management.tasks',
])
app.conf.update(
CELERY_IGNORE_RESULT=True,
CELERY_ACCEPT_CONTENT=['json'],
CELERY_TASK_SERIALIZER='json',
CELERYBEAT_SCHEDULE={
'periodic_task': {
'task': 'task_polling',
'schedule': timedelta(seconds=1),
},
},
)
我使用以下命令启动这些工作人员：
celery -A my_app worker -B -l info
在启动屏幕上，任务已很好地注册：
-------------- celery@mbp-de-julio.home v3.1.12 (Cipater)
---- **** -----
--- * *** * -- Darwin-13.1.0-x86_64-i386-64bit
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app: __main__:0x102341390
- ** ---------- .> transport: amqp://guest:**@localhost:5672//
- ** ---------- .> results: disabled
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ----
--- ***** ----- [queues]
-------------- .> celery exchange=celery(direct) key=celery
[tasks]
. save_shell_task
. task_polling
另一方面，我这样定义任务和 celery 配置：
from celery import Celery
from celery import shared_task
from celery.execute import send_task
from task_workers import task_workers
app = Celery('tasks', broker='amqp://guest@localhost//')
@shared_task(name='do_work')
def do_work(_serialized_task):
result = task_workers.Worker().do(_task=_serialized_task)
send_task('save_shell_task', [], {'_result':result})
我用这个命令启动这个工人：
celery -A tasks worker --loglevel=info
开始屏幕显示任务已成功注册：
-------------- celery@mbp-de-julio.home v3.1.12 (Cipater)
---- **** -----
--- * *** * -- Darwin-13.1.0-x86_64-i386-64bit
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app: tasks:0x101833450
- ** ---------- .> transport: amqp://guest:**@localhost:5672//
- ** ---------- .> results: disabled
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ----
--- ***** ----- [queues]
-------------- .> celery exchange=celery(direct) key=celery
[tasks]
. do_work
但是好像celery的两个实例不知道怎么交流：
[2014-07-09 22:21:52,184: ERROR/MainProcess] Received unregistered task of type u'task_polling'.
The message has been ignored and discarded.
Did you remember to import the module containing this task?
Or maybe you are using relative imports?
Please see http://bit.ly/gLye1c for more information.
The full contents of the message body was:
[...]
KeyError: u'task_polling'
如何在双方配置 celery 以解决通信问题？
10/07 更新
当我单独使用轮询任务启动节拍时，轮询正在工作，但是当我与其他工作人员一起启动 celery 时，此 celery 收到轮询消息并说他不知道此任务。也许我必须为任务分配不同的队列？
我定义了队列。
在工人类型 2 方面，celeryconfig.py 文件：
CELERY_ROUTES = {
'do_work': {
'queue': 'task_workers_queue'
},
'task_polling': {
'queue': 'db_workers_queue'
},
'save_shell_task': {
'queue': 'db_workers_queue'
},
}
在 db worker 类型 1 方面：
app.conf.update(
CELERY_IGNORE_RESULT=True,
CELERY_ACCEPT_CONTENT=['json'],
CELERY_TASK_SERIALIZER='json',
CELERYBEAT_SCHEDULE={
'periodic_task': {
'task': 'task_polling',
'schedule': timedelta(seconds=1),
},
},
CELERY_ROUTES={
'task_polling': {
'queue': 'db_workers_queue'
},
'save_shell_task': {
'queue': 'db_workers_queue'
},
'do_work': {
'queue': 'task_workers_queue'
}
},
)
我这样启动芹菜：
$ celery -A tasks worker --loglevel=info -Q task_workers_queue
$ celery -A ussd_auto worker -B -l info -Q db_workers_queue
现在周期性任务db_polling 很好地被节拍调用了。
此任务成功地将作业发送到do_work 任务。
但是do_work 任务未能将结果发送到任务save_shell_task。
没有错误但没有附加任何内容。
【问题讨论】：