背景
我们团队有很多的前端热更新包需要频繁的更新到外网,一般会有上百兆的包上传到各个地区(国内,台湾 ,海外等区域)。刚开始是手动上传,通过中间几层的中转才能到指定目的地,经常网络中断,再重传,而且需要专人等候以便应对无法预料的问题,很耗时耗力。通过调研某云厂商有对象存储服务,能够支持本地上传,能支持在任何区域下载。这就解决了我们中间中转问题。然后找到官方文档,通过调用对象存储的API上传更新包,确实方便了很多,但是上传一个大文件要等几十分钟,这就比较累了,还是得等待。只能继续找解决方案了,通过调研,Celery能支持异步任务处理,只需要给发起任务给celery,然后celery处理任务,执行完任务后通过回调通知,这样我们就只需要关注最终执行结果就行了。另外这个方案也解决了支持多任务处理,之前的方案不能支持并发处理。现在团队只需要将更新包放在指定位置,然后触发上传任务并填写个人邮箱地址,后续工作交给celery处理,处理完后邮件通知下载地址,这样是不是简单多了。下面我们来了解下celery。
Celery简介
Celery是一个简单,可靠,灵活的分布式系统,用于处理大量的消息。Celery是致力于实时处理的任务队列,并且支持任务调度。Celery是python语言编写,但它的协议可供其他语言使用。
Celery的优点:
简单:Celery容易使用和维护,不需要配置文件。
高可用:工作进程和客户端连接断开或者连接失败能够自动重连,并且和一些代理支持主主、主从类型的高可用。
快:单个celery进程能够每秒处理百万级的任务,使用broker和优化的配置能达到小于毫秒的延迟。
灵活:Celery的每个部分都能被扩展或者单独使用,能够客户化pool,序列化,压缩机制,日志记录,调度,消费者,生产者,传输代理等。
Celery架构
简单说明上面的基本流程:
Client A发送任务请求给broker;
broker收到请求并存放在任务队列;
如果celery没有空闲的worker则等待,如果这时celery有空闲的worker,则获取任务并执行;
Celery的 worker执行完任务后将执行结果写到Backend results;
Client A通过定时轮询查询执行结果;
补充:
Celery 的broker可支持:RabbitMQ,Redis,Amazon SQS等
Celery 的Result Stores可支持:Redis,Memcached,AMQP,Django ORM,Elasticsearch等
异步上传功能架构图
上传功能的简单流程:
用户通过浏览器填写相关信息,发起文件上传请求;
Django将上传任务下发给redis;
Celery worker获取任务并执行任务;
Worker执行完任务并回调发送邮件功能,同时将执行的结果写入redis;
邮件服务器将邮件分发给指定收件人,含下载地址;
核心代码演示:
安装celery
pip install celery[redis]
编写celery worker代码:
将下面内容保存为:testtasks.py
#!/usr/bin/python
# coding:utf-8
from celery import Task
from celery import Celery
app=Celery('test',broker='redis://127.0.0.1/3',backend="redis://127.0.0.1/4")
class CallbackTask(Task):
def on_success(self, retval, task_id, args, kwargs):
'''成功执行的函数'''
print "callback success function"
def on_failure(self, exc, task_id, args, kwargs, einfo):
'''失败执行的函数'''
print "callback failure function"
@app.task(base=CallbackTask)
def add(x, y):
return x + y
命令行启动:
celery -A testtasks worker --loglevel=info
调用testtasks任务:
>>> from testtasks import add
>>> add.delay(3,4)
<AsyncResult: 75cccb61-dd47-4577-82ab-cc1fe55d5a0a>
查看执行结果:
查看redis db 4保存的执行的结果:
常见问题:
再实验下celery worker异常中断,会出现什么情况?
通过切换到redis db 3发现有任务未执行,内容如下,这就说明整个异步处理的高可用方案只需要保障broker的高可用,正常启动celery worker后会自动获取任务执行;
你的任务被多次执行?
假如一个任务没有确认在visibility_timeout时间内,这个将会被重新分配到其他worker并执行。默认visibility_timeout被设置为1小时,如果你确定有任务会超过一小时,建议设置为1天。
BROKER_TRANSPORT_OPTIONS = {'visibility_timeout': 86400} #1 day
最佳实践
Broker和backend选择:在生产环境中,broker最好使用Redis和RabbitMQ,backend最好使用memcached或者Redis。
Visibility Timeout设置:如果你的任务有超过1小时的,考虑把这个时间设置更大。
使用优先级的队列:如果你有非常多的任务并且量非常大,请使用优先级队列吧。例如我们的抢票软件,能够根据你优先级给你排队抢票。
尽量保持任务的简洁:不要把你的业务逻辑放在里面,里面放核心功能,比如生成报表,发送邮件等;
监控你的celery:使用flower,下面是安装方法以及使用。
pip install [https://github.com/mher/flower/zipball/master](https://github.com/mher/flower/zipball/master)
celery flower -A testtasks --address=0.0.0.0 --port=5555
下面可用于自写脚本启动(多进程):
启动celery worker
celery multi start -A testtasks worker --pidfile="logs/worker.pid" --logfile="logs/worker.log"
关闭celery worker
celery multi kill --pidfile="logs/worker.pid"
参考文档
https://khashtamov.com/en/celery-best-practices-practical-approach
https://github.com/mher/flower