• 如果您觉得本站非常有看点,那么赶紧使用Ctrl+D 收藏吧

基于Celery在多台云服务器上实现分布式

开发技术 开发技术 3周前 (09-05) 24次浏览

起源

最近参加公司里的一个比赛,比赛内容里有一项是尽量使用分布式实现项目。因为项目最终会跑在jetsonnano,一个贼卡的开发板,性能及其垃圾。而且要求使用python
找了很多博客,讲的真的是模棱两可,最后结合官方文档终于啃出来,写出来分享一下。

目前本博客的几个要点:

  1. 使用celery实现分布式
  2. 结合flask框架使用
  3. 使用Redis作为broker
  4. 使用Redis作为backend

准备工作

安装环境

需要安装的环境包括Redis、Celery、Pyhton3.6.9「开发板自带」

这里需要填坑的是Redis的设置。因为在腾讯云服务器、华为云服务器、滴滴云服务器上做的测试「穷,每个平台褥点羊毛」,一开始不知道在安全组里面开放端口,一直连接不上服务器,很坑。这里如果没有使用云服务器的话可以跳过安全组这一步的设置,使用了云服务器的话,一件开放一下端口,自行百度开放方法。

首先是Redis的设置,Redis需要把默认的120.0.0.1 IP地址修改为0.0.0.0,并把守护进程关闭。

vim redis.conf
bind = 0.0.0.0
protected -mode no //从yes改为no

安装celery

pip3 install celery -i https://pypi.tuna.tsinghua.edu.cn/simple

一个简单的分布式Demo

网上关于创建Celery文件的描述都很模糊,这里我的理解是这样,首先看一下官方给出的Demo:

from celery import Celery

app = Celery('tasks', broker='pyamqp://guest@localhost//')

@app.task
def add(x, y):
    return x + y

将这个文件命名为tasks.py

这段代码最关键的核心是app = Celery('tasks', broker='pyamqp://guest@localhost//')
既然是分布式,肯定要有worker,干活的人,也就是云服务器,在云服务器上需要做的事情是:celery -A tasks worker --loglevel=info
其中的celery -A *** worker --loglevel=info *** 就是worker要拿到任务的任务板标志,只有有个这个标志,worker才知道到底谁在发任务。

broker='pyamqp://guest@localhost//' 因为我使用redis来作为任务的存放容器,所以改为 broker='redis://guest@localhost//'
broker是存放任务的地方,所以我把发放任务的服务器的地址填进去: app = Celery('tasks', broker='redis://121.***.***.190:6379')
6379为默认的端口号,其实broker这段url应该包括redis的用户名、用户密码+IP地址。因为我们前面修改的redis的配置文件,所以这里可以无密码访问。

@app.task
def add(x, y):
    return x + y

这一段就是服务器要发送出去的任务了。当然服务器里不需要包含执行任务所需要的库,库安装在worker的服务器里就可以了。「当然add(x,y)啥库也不需要」。

现在可以来见识一下celery的威力了,把上面修改后的tasks.py放到worker服务器上面,执行命令:celery -A tasks worker --loglevel=info
你会看到下面这行:

(base) zhaosi@zhaosideMBP *** % celery -A tasks worker --loglevel=info
 
 -------------- celery@zhaosideMBP v4.4.7 (cliffs)
--- ***** ----- 
-- ******* ---- macOS-10.15.6-x86_64-64bit 2020-09-05 14:35:13
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x7fd0bb1b16a0
- ** ---------- .> transport:   redis://121.***.***.190:6379/8
- ** ---------- .> results:     redis://121.***.***.190:6379/7
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                
[tasks]
  . tasks.add

[2020-09-05 14:35:15,566: INFO/MainProcess] Connected to redis://121.***.***.190:6379/8
[2020-09-05 14:35:15,773: INFO/MainProcess] mingle: searching for neighbors
[2020-09-05 14:35:17,484: INFO/MainProcess] mingle: all alone
[2020-09-05 14:35:18,789: INFO/MainProcess] celery@zhaosideMBP ready.

当你看到最后四行时,YES,最简单的Demo被你跑起来了

[tasks]
  . tasks.add

这里展示的是worker可以接到的任务,当然现在服务器还没有发布任务,worker在持续监听服务器上存储发布任务的redis数据库,等着接活。

验收成果

打开云服务器,准备发布任务「请在服务器上也创建一个tasks.py,不需要安装任何依赖」

root@-x:~/pro# python3
Python 3.6.9 (default, Jul 17 2020, 12:50:27) 
[GCC 8.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from tasks import add
>>> add.delay(1,1)
<AsyncResult: 71c49a0e-2d52-444e-b158-1ae9f1486767>
>>> 

回到worker服务器,可以看到任务被接收并且完成了!

[2020-09-05 14:41:10,138: INFO/MainProcess] Received task: tasks.add[71c49a0e-2d52-444e-b158-1ae9f1486767]  
[2020-09-05 14:41:10,223: WARNING/ForkPoolWorker-8] 2

可以通过result来接受结果:

>>> result = add.delay(1,1)
<AsyncResult: 71c49a0e-2d52-444e-b158-1ae9f1486767>
>>> result.get()
2

更复杂的返回值请各位自行探索啦
所谓万事开头难,有了这个Demo的帮助,后续的任务会简单很多。

后续

还是官方文档好啊。现在大多数博客写的是个啥。


喜欢 (0)