Celery分布式实践

快来打我* 2021-09-21 00:14 360阅读 0赞

> Celery3.1.7文档：[http://docs.jinkan.org/docs/celery/index.html][http_docs.jinkan.org_docs_celery_index.html]；
> 
> Celery4.4.0文档：[http://docs.celeryproject.org/en/master/index.html][http_docs.celeryproject.org_en_master_index.html]

基本的Celery执行任务分发实现在我的另一篇博文“[Celery学习记录][Celery]”已有描述，这里不再赘述。

# 一、基本文件的创建与配置 #

这里配置在windows或者是linux中均可。甚至可以有的节点为windows有的节点为linux，不冲突。

![20191205221828163.png][]

上图是我的项目的文件路径及项目包含的文件："celery\_config.py，celery.py， \_\_init\_\_.py， tasks.py"。

（1）\_\_init\_\_.py

用于标识当前文件夹为一个python package。文件内容为空。

（2）celery.py

用于定义app：

# -*- coding:utf-8 -*-
    from __future__ import absolute_import
    from celery import Celery
    
    app = Celery('proj', )   # include=['proj.tasks']
    app.config_from_object('proj.celery_config')
    
    
    if __name__ == "__main__":
        app.start()

（3）celery\_config.py

配置文件。

# -*- coding:utf-8 -*-
    # from __future__ import absolute_import
    from kombu import Exchange, Queue
    
    BROKER_URL = 'redis://****:6379/3'  # broker
    CELERY_RESULT_BACKEND = 'redis://*****:6379/4'  # 存储任务状态和结果
    
    CELERY_IMPORTS = ('proj.tasks', )    # 开始时要导入的模块列表
    
    CELERY_TASK_SERIALIZER = 'json'  # 任务序列化和反序列化    pickle
    
    CELERY_RESULT_SERIALIZER = 'json'  # 结果序列化格式，默认为pickle
    
    CELERY_ACCEPT_CONTENT = ['json']  # 指定接受的内容类型(默认为允许所有格式)   ['pickle', 'json', 'msgpack', 'yaml']
    CELERY_ENABLE_UTC = True
    CELERY_TIMEZONE = 'Asia/Shanghai'
    
    CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24  # 任务过期时间
    
    # ---------------------定义路由、交换、队列------------------------
    
    CELERY_DEFAULT_QUEUE = 'default'
    CELERY_QUEUES = (
        Queue('default', Exchange('default'), routing_key='default'),
        Queue('add', Exchange('add'), routing_key='add'),
        Queue('multi', Exchange('multi'), routing_key='multi'),
    )
    CELERY_DEFAULT_EXCHANGE = 'default'
    CELERY_DEFAULT_EXCHANGE_TYPE = 'direct'
    CELERY_DEFAULT_ROUTING_KEY = 'default'

其中，BROKER\_URL、CELERY\_RESULT\_BACKEND分别对应于你实际使用的中间人broker与结果存储后端backend，我这里使用的是redis。CELERY\_QUEUES中包含的Queue为定义的队列，目的是用于实现在不同的节点上执行不同的任务，如果不需要区分不同的节点，则无需设置与Queue、Exchange、Routing相关的项。具体这几项设置的含义，参见[Routing Tasks][]（文档）。

（4）tasks.py

定义用于执行的任务：

# -*- coding:utf-8 -*-
    from __future__ import absolute_import
    import time
    from proj.celery import app
    
    
    @app.task
    def add(x, y):
        print "this is the add function..."
        time.sleep(1)
        return x+y
    
    
    @app.task
    def multi(x, y):
        print "this is the multi function..."
        time.sleep(1)
        return x*y

**最后，复制上述代码到另一个服务器节点，这里我的另一个节点为windows环境。 **

理论上，可以有任意多个节点。

# 二、启动Broker及Backend #

broker及backend能够正常运行、通信，是celery运行的必须条件。

这里我用的是redis，下载安装及配置在博文[Celery学习记录][Celery]都有，不再赘述。

根据配置进行启动： /usr/local/redis-4.0.11/src/redis-server    /usr/local/redis-4.0.11/redis.conf

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70][]

# 三、发布任务 #

## 1、简单实现分布式，不区分节点执行的任务内容 ##

首先，在linux(centos)下，执行 celery -A proj worker -l info  启动任务监听。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 1][]

在windows下启动任务监听：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 2][]

可以看到后启动的windows监视窗口，输出显示有一个neighbors节点，即我们先启动的linux节点。这时，你再去观察先开启的linux下的窗口同样出现了![20191205224916286.png][]

类似的消息，表示有新的节点加入到了集群中。

现在，可以在python环境，或者写一个py文件来执行文件，这里我写了一个py文件，如下：

# -*- coding:utf-8 -*-
    from proj.tasks import add, multi
    
    
    for i in range(10):
        re1 = add.apply_async(args=[i, i])   # 调用add task
        re2 = multi.apply_async(args=[i, i])  # 调用 multi task
        re1.get()
        re2.get()

注意，这个py文件应该放在proj 的同级文件夹下，不是proj中。这里，我是循环调用了add和multi方法。可以在windows或者linux任意一个节点中run均可"**python 文件名.py**"，不妨这里就叫**a.py**。

效果如下：

linux监视窗口：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 3][]

windows环境下的监视窗口：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 4][]

不难发现，两个节点均执行了add和multi函数，没有加以区分，并且混合出现。

## 2、每个节点执行指定队列的任务 ##

只需要在启动Celery服务时，加上“-Q”参数（前提是已经在配置文件celery\_config.py中定义好了队列），-Q参数指明了当前节点要监视、执行的任务队列。

![20191206090545795.png][]

根据定义好的队列，启动Celery服务：

（1）启动linux节点的Celery服务：

celery -A proj worker -l info **-Q add**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 5][]

注意，这里启动后console中显示的输出日志，包含了queues，其内容与我们定义及启动的队列相一致。

（2）启动windows节点的Celery服务：

celery -A proj worker -l info -Q multi

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 6][]

（3）发布任务

修改我们先前写的a.py：

# -*- coding:utf-8 -*-
    from proj.tasks import add, multi
    
    
    for i in range(10):
        re1 = add.apply_async(args=[i, i], queue='add', routing_key='add')
        re2 = multi.apply_async(args=[i, i], queue='multi', routing_key='multi')
        #  或者写为：re2 = multi.apply_async(kwargs={'x': i, 'y': i}, queue='multi', routing_key='multi')
        re1.get()
        re2.get()

注意，发布任务时，指明了任务所要发布到的队列及队列的密匙(参见：[Routing Tasks][])。

观察linux及windows节点的Celery监控日志，发现：linux节点只执行了add任务，windows节点只执行了multi任务，与我们发布的任务队列一致。

linux：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 7][]

windows：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 8][]

注意：在配置文件中我还定义了一个“default”队列，发布任务时并没有用到，当然这个队列可以选择去掉或者保留。并且，每一个节点可以执行的任务队列可以为多个只需在-Q参数后用英文 “,” 隔开即可。例如：celery -A proj worker -l info -Q multi, default。这个default队列的另一个作用是，当有任务没有被指定队列时，则默认分配到default队列中。

## 3、通过配置文件将任务绑定到指定的队列 ##

修改配置文件 **celery\_config.py :**

# -*- coding:utf-8 -*-
    # from __future__ import absolute_import
    from kombu import Exchange, Queue
    
    BROKER_URL = 'redis://****:6379/3'  # broker
    CELERY_RESULT_BACKEND = 'redis://*****:6379/4'  # 存储任务状态和结果
    
    CELERY_IMPORTS = ('proj.tasks', )    # 开始时要导入的模块列表
    
    CELERY_TASK_SERIALIZER = 'json'  # 任务序列化和反序列化    pickle
    
    CELERY_RESULT_SERIALIZER = 'json'  # 结果序列化格式，默认为pickle
    
    CELERY_ACCEPT_CONTENT = ['json']  # 指定接受的内容类型(默认为允许所有格式)   ['pickle', 'json', 'msgpack', 'yaml']
    CELERY_ENABLE_UTC = True
    CELERY_TIMEZONE = 'Asia/Shanghai'
    
    CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24  # 任务过期时间
    
    # ---------------------定义路由、交换、队列------------------------
    # http://docs.jinkan.org/docs/celery/userguide/routing.html
    CELERY_ROUTES = ({'proj.tasks.add': {
                            'queue': 'add',
                            'routing_key': 'add'
                     }}, )
    
    CELERY_DEFAULT_QUEUE = 'default'
    CELERY_QUEUES = (
        Queue('default', Exchange('default'), routing_key='default'),
        Queue('add', Exchange('add'), routing_key='add'),
        Queue('multi', Exchange('multi'), routing_key='multi'),
    )
    CELERY_DEFAULT_EXCHANGE = 'default'
    CELERY_DEFAULT_EXCHANGE_TYPE = 'direct'
    CELERY_DEFAULT_ROUTING_KEY = 'default'

注意，此部分修改的内容为增加了CELERY\_ROUTES参数：![20191209111800396.png][]

修改发布任务的**a.py**:![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 9][]

即，将add函数的queue及routing\_key去掉（使用配置文件的绑定设置）。

重启各节点的Celery服务，并在任意一个节点运行python a.py。观察结果：

（1）使用celery -A proj worker -l info **-Q add** 启动的节点：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 10][]

（2）使用 celery -A proj worker -l info **-Q multi **启动的节点：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 11][]

此部分与上部分“2”很相近。不同的是“2”是指定了几个队列，每个节点监听不同的队列，任务分发的时候需要将任务通过

apply_async(args=, queue='', routing_key='')

方法指定到特定的队列中去。而本部分，可以通过直接将任务绑定到指定的任务队列，任务发布时无需再指定队列，即使用 apply\_async 方法时，不需要queue及routing\_key参数。

# 附： #

可以在浏览器窗口查看任务执行情况，方法是在开启新的命令窗口执行“celery flower --broker=你的实际broker设置”，然后根据输出提示，访问http://localhost:5555，当然也可以使用“--port=5555” 指定端口。方法参见：[Monitoring and Management Guide][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 12][]

效果图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 13][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 14][]

> 推荐阅读（其实最好还是去看一遍官方文档）：
> 
> [celery\#集群管理实现][celery]；
> 
> [CELERY 集群管理实现][CELERY]

[http_docs.jinkan.org_docs_celery_index.html]: http://docs.jinkan.org/docs/celery/index.html
[http_docs.celeryproject.org_en_master_index.html]: http://docs.celeryproject.org/en/master/index.html
[Celery]: https://blog.csdn.net/qq_27378621/article/details/102979718
[20191205221828163.png]: /images/20210920/567404eb60354d00840bd6e1a7e40035.png
[Routing Tasks]: http://docs.jinkan.org/docs/celery/userguide/routing.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70]: /images/20210920/bae549e79d27400e88445c92003e789e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 1]: /images/20210920/481741a86dbf471db535f9e37f4f3f5e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 2]: /images/20210920/cbc78c50ee4544bdb6752fd6fa37687f.png
[20191205224916286.png]: /images/20210920/ff8e4c3e5b254d2f899f3afdbdaee745.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 3]: /images/20210920/45537e617d47494aafbf42bbb5e77c2f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 4]: /images/20210920/902819b6fcd041d291063ddfd03ac046.png
[20191206090545795.png]: /images/20210920/5810cddda5e647bf89759736facdadca.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 5]: /images/20210920/30f07c965aae443eb8c23e66330fbcf1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 6]: /images/20210920/e46ed0b9f18140d49b13c0d394c73d42.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 7]: /images/20210920/a237515a4ed347619c0eacc996d09905.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 8]: /images/20210920/7743135498b041f48bd394472b3eabe4.png
[20191209111800396.png]: /images/20210920/06d4e02991d14decabd8ea1f69ceb74f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 9]: /images/20210920/d80e591e25884f7c9bc934133f664b1e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 10]: /images/20210920/5ab9c556c0b14838942a58bcf5b392a4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 11]: /images/20210920/989bea95d1ea467c83cf1540f02e9f29.png
[Monitoring and Management Guide]: http://docs.jinkan.org/docs/celery/userguide/monitoring.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 12]: /images/20210920/472c8b1effd441098013957d7e1f3e54.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 13]: /images/20210920/3d4b1c0feca44343952594fe3efb2f36.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3Mzc4NjIx_size_16_color_FFFFFF_t_70 14]: /images/20210920/2772f66a2c1443efb7128e000dc1fa7c.png
[celery]: https://my.oschina.net/hochikong/blog/518587
[CELERY]: https://www.jianshu.com/p/a7d453abed3c