Celery手动配置路由

 

1、需求

我们有多种不同的任务，这些任务优先级不同，比如我们有视频上传和压缩任务等，照片压缩上传等任务还有其他不重要的任务。这些任务耗时不同需要使用不同的worker去处理。只是用celery默认的队列就不能满足我们的需求了。
这就需要我们将不同的task路由到不同队列，让不同的worker处理不同种类的task。

2、创建队列和交换机

关于交换机和队列可以先看看http://rabbitmq.mr-ping.com/

default_exchange = Exchange('dedfault', type='direct')

# 定义一个媒体交换机,类型是直连交换机
media_exchange = Exchange('media', type='direct')

# 创建三个队列，一个是默认队列，一个是video、一个image
CELERY_QUEUES = (
    Queue('default', default_exchange, routing_key='default'),
    Queue('videos', media_exchange, routing_key='media.video'),
    Queue('images', media_exchange, routing_key='media.image')
)
# 定义默认队列和默认的交换机routing_key
CELERY_DEFAULT_QUEUE = 'default'
CELERY_DEFAULT_EXCHANGE = 'default'
CELERY_DEFAULT_ROUTING_KEY = 'default'

3、创建task

from celery import Celery
import time

app = Celery()
app.config_from_object('celeryconfig')

# 视频压缩
@app.task
def video_compress(video_name):
    time.sleep(10)
    print 'Compressing the:', video_name
    return 'success'

@app.task
def video_upload(video_name):
    time.sleep(5)
    print u'正在上传视频'
    return 'success'

# 压缩照片
@app.task
def image_compress(image_name):
    time.sleep(10)
    print 'Compressing the:', image_name
    return 'success'

# 其他任务
@app.task
def other(str):
    time.sleep(10)
    print 'Do other things'
    return 'success'

我们已经定义了三个队列，现在我们想将操作视频和操作照片的task分别路由到特定的队列。

4、指定路由

CELERY_ROUTES = ({'tasks.image_compress': {
                        'queue': 'images',
                        'routing_key': 'media.image'
                 }},{'tasks.video_upload': {
                        'queue': 'videos',
                        'routing_key': 'media.video'
                 }},{'tasks.video_compress': {
                        'queue': 'videos',
                        'routing_key': 'media.video'
                 }}, )

通过CELERY_ROUTES来为每一个task指定队列，如果有任务到达时，通过任务的名字来让指定的worker来处理。

5、task注册

关于任务的名字可以看看这篇文档http://docs.jinkan.org/docs/celery/userguide/tasks.html
celery可以自动生成名字，如果任务没有注册，就会出错。搜索后发现有人使用下面方法解决。

CELERY_IMPORTS = ("tasks",)

tasks是我保存任务的模块名，这样在创建worker时就可以将任务注册到worker，如下面这样：

 

图片中的[tasks]下面的几个任务就是我tasks文件中的任务。

6、完整代码

tasks.py

from celery import Celery
import time


app = Celery()
app.config_from_object('celeryconfig')

# 视频压缩
@app.task
def video_compress(video_name):
    time.sleep(10)
    print 'Compressing the:', video_name
    return 'success'

@app.task
def video_upload(video_name):
    time.sleep(5)
    print u'正在上传视频'
    return 'success'

# 压缩照片
@app.task
def image_compress(image_name):
    time.sleep(10)
    print 'Compressing the:', image_name
    return 'success'

# 其他任务
@app.task
def other(str):
    time.sleep(10)
    print 'Do other things'
    return 'success'

celeryconfig.py

from kombu import Exchange, Queue
from routers import MyRouter

# 配置市区
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_BROKER = 'amqp://localhost'
# 定义一个默认交换机
default_exchange = Exchange('dedfault', type='direct')

# 定义一个媒体交换机
media_exchange = Exchange('media', type='direct')

# 创建三个队列，一个是默认队列，一个是video、一个image
CELERY_QUEUES = (
    Queue('default', default_exchange, routing_key='default'),
    Queue('videos', media_exchange, routing_key='media.video'),
    Queue('images', media_exchange, routing_key='media.image')
)

CELERY_DEFAULT_QUEUE = 'default'
CELERY_DEFAULT_EXCHANGE = 'default'
CELERY_DEFAULT_ROUTING_KEY = 'default'
#
CELERY_ROUTES = ({'tasks.image_compress': {
                        'queue': 'images',
                        'routing_key': 'media.image'
                 }},{'tasks.video_upload': {
                        'queue': 'videos',
                        'routing_key': 'media.video'
                 }},{'tasks.video_compress': {
                        'queue': 'videos',
                        'routing_key': 'media.video'
                 }}, )

# 在出现worker接受到的message出现没有注册的错误时，使用下面一句能解决
CELERY_IMPORTS = ("tasks",)

注意在启动worker的时候需要制定队列，需要在保存任务的目录中打开终端启动worker，这个问题我还没有搞明白！

# 启动worker1
celery worker -Q default --loglevel=info

celery worker -Q videos --loglevel=info

启动处理图片的worker

celery worker -Q images --loglevel=info

这样我们就可以把不同类的任务路由到不同的worker上处理了。

遇到的坑

在你修改了配置文件需要重启worker时，记得把python shell也关掉重启。
如果调用任务但是worker出现这样的错误

unregistered task of type

就说明你的任务没有注册，需要加上我上面提到的CELERY_IMPORTS = ("tasks",)到配置文件中。
建议将还是自己多动手调试，把芹菜的官方文档看看，里面的东西会让我们看了豁然开朗。

作者：嘿嘿_小余同学
链接：http://www.jianshu.com/p/11b420aea529
來源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。