笔记：集群部署celery分布式任务队列

柔光的暖阳◎ 2023-07-19 10:56 24阅读 0赞

[https://www.213.name/archives/1105][https_www.213.name_archives_1105]

[https://zhuanlan.zhihu.com/p/100109543][https_zhuanlan.zhihu.com_p_100109543]

[https://github.com/SpiderClub/weibospider/wiki/Celery%E9%85%8D%E7%BD%AERedis-Sentinel%E5%81%9A%E9%AB%98%E5%8F%AF%E7%94%A8][https_github.com_SpiderClub_weibospider_wiki_Celery_E9_85_8D_E7_BD_AERedis-Sentinel_E5_81_9A_E9_AB_98_E5_8F_AF_E7_94_A8]

[https://rookiefly.cn/detail/168][https_rookiefly.cn_detail_168]  
  
估计来看此文的已经对celery有一些了解了，基本概念不再赘述。

网上找来找去全都是单机版的基础部署教程，也没有深入讲解分布式的部署过程。没办法只好靠着我渣渣英语强行研究了一波官方文档。

只有三台阿里云的服务器，不过做集群演示是够了。服务器配置比较低，所以选用redis作为消息中间件。

准备工作  
实验环境：centos 6.5，python 3.4

三台服务器都装上python 3.4和celery 4.1

我pip使用的阿里云的源，默认celery版本是4.1.0，直接使用sudo pip install celery即可安装

其中一台做主服务器装上redis和mongodb，和celery的flower插件

redis做broker

mongodb用来做backend，存储任务执行结果

flower用来监测celery集群的状态

安装教程请自行搜索

注：  
celery命令的详细参数请看官方文档：http://docs.celeryproject.org/en/master/reference/celery.bin.worker.html

从celery 4.0开始配置文件有所变化，具体请看官方文档：http://docs.celeryproject.org/en/master/userguide/configuration.html\#example-configuration-file

queue配置：http://blog.csdn.net/tmpbook/article/details/52245716

编写代码  
项目结构：

init.py

from celery import Celery

app = Celery("demo")  
app.config\_from\_object("celery\_app.celeryconfig")  
Python  
celeryconfig.py

from celery.schedules import crontab  
from datetime import timedelta  
from kombu import Queue  
from kombu import Exchange

result\_serializer = 'json'

broker\_url = "redis://192.168.1.2"  
result\_backend = "mongodb://192.168.1.2/celery"  
timezone = "Asia/Shanghai"  
imports = (  
    'celery\_app.task1',  
    'celery\_app.task2'  
)

beat\_schedule = \{  
    'add-every-20-seconds': \{  
        'task': 'celery\_app.task1.multiply',  
        'schedule': timedelta(seconds=20),  
        'args': (5, 7)  
    \},  
    'add-every-10-seconds': \{  
        'task': 'celery\_app.task2.add',  
         'schedule': crontab(hour=9, minute=10)  
        'schedule': timedelta(seconds=10),  
        'args': (23, 54)  
    \}  
\}

task\_queues = (  
    Queue('default', exchange=Exchange('default'), routing\_key='default'),  
    Queue('priority\_high', exchange=Exchange('priority\_high'), routing\_key='priority\_high'),  
    Queue('priority\_low', exchange=Exchange('priority\_low'), routing\_key='priority\_low'),  
)

task\_routes = \{  
    'celery\_app.task1.multiply': \{'queue': 'priority\_high', 'routing\_key': 'priority\_high'\},  
    'celery\_app.task2.add': \{'queue': 'priority\_low', 'routing\_key': 'priority\_low'\},  
\}

\# 每分钟最大速率  
\# task\_annotations = \{  
\#     'task2.multiply': \{'rate\_limit': '10/m'\}  
\# \}

Python  
task1.py

import time  
from celery\_app import app

@app.task  
def multiply(x, y):  
    print("multiply")  
    time.sleep(4)  
    return x \* y  
Python  
task2.py

import time  
from celery\_app import app

@app.task  
def add(x, y):  
    print(add)  
    time.sleep(2)  
    return x + y  
Python  
上传项目  
写个同步文件sync.sh，方便将项目文件分别上传到三台服务器

scp -r celery\_app root@192.168.1.2:~  
scp -r celery\_app root@192.168.1.3:~  
scp -r celery\_app root@192.168.1.4:~  
\#开始上传  
chmod +x sync.sh  
./sync.sh  
Bash  
其实不是每台服务器都需要所有的文件，只是为了方便就全部上传上去了

启动  
\#worker 1  
celery -A celery\_app -l info -n worker1  
\#worker 2  
celery -A celery\_app -l info -n worker2  
\#broker  
/var/mongodb/bin/mongod -f /var/mongodb/conf/mongod.conf

nohup redis-server &

nohup celery -A celery\_app flower -l info &

celery -A celery\_app beat -l info

Bash  
全部启动完成之后可以看到celery已经开始自动分配定时任务了

提交自定义任务  
test.py

from celery\_app import task1  
from celery\_app import task2

re = task1.multiply.delay(2, 8)  
re2 = task2.add.delay(5, 6)  
print("ok")  
print(re.get())  
print(re2.get())  
Python  
输出：

/usr/bin/python3.4 /media/chengyu/Project/Github/CMSpider/test.py  
ok  
16  
11

Process finished with exit code 0  
Null  
进入flower监控服务器

可能遇到的错误  
pymongo.errors.NotMasterError: not master

关闭mongodb的数据集

其他  
当服务器数量较多的时候，管理起来会很不方便，可以使用python的supervisor来管理后台进程，遗憾的是它并不支持python3，不过也可以装在python2的环境

虽然用了supervisor可以很方便的管理python程序，但是还是得一个个登陆不同的服务器的去管理，咋办捏？

我在github上找到一个工具supervisor-easy，可以批量管理supervisor，如图:

地址：https://github.com/trytofix/supervisor-easy

\`(\*∩\_∩\*)′转载请注明：转载自：笔记：集群部署celery分布式任务队列

[https_www.213.name_archives_1105]: https://www.213.name/archives/1105
[https_zhuanlan.zhihu.com_p_100109543]: https://zhuanlan.zhihu.com/p/100109543
[https_github.com_SpiderClub_weibospider_wiki_Celery_E9_85_8D_E7_BD_AERedis-Sentinel_E5_81_9A_E9_AB_98_E5_8F_AF_E7_94_A8]: https://github.com/SpiderClub/weibospider/wiki/Celery%E9%85%8D%E7%BD%AERedis-Sentinel%E5%81%9A%E9%AB%98%E5%8F%AF%E7%94%A8
[https_rookiefly.cn_detail_168]: https://rookiefly.cn/detail/168