背景
我们团队有很多的前端热更新包需要频繁的更新到外网，一般会有上百兆的包上传到各个地区（国内，台湾 ，海外等区域）。刚开始是手动上传，通过中间几层的中转才能到指定目的地，经常网络中断，再重传，而且需要专人等候以便应对无法预料的问题，很耗时耗力。通过调研某云厂商有对象存储服务，能够支持本地上传，能支持在任何区域下载。这就解决了我们中间中转问题。然后找到官方文档，通过调用对象存储的API上传更新包，确实方便了很多，但是上传一个大文件要等几十分钟，这就比较累了，还是得等待。只能继续找解决方案了，通过调研，Celery能支持异步任务处理，只需要给发起任务给celery，然后celery处理任务，执行完任务后通过回调通知，这样我们就只需要关注最终执行结果就行了。另外这个方案也解决了支持多任务处理，之前的方案不能支持并发处理。现在团队只需要将更新包放在指定位置，然后触发上传任务并填写个人邮箱地址，后续工作交给celery处理，处理完后邮件通知下载地址，这样是不是简单多了。下面我们来了解下celery。
Celery简介
Celery是一个简单，可靠，灵活的分布式系统，用于处理大量的消息。Celery是致力于实时处理的任务队列，并且支持任务调度。Celery是python语言编写，但它的协议可供其他语言使用。
Celery的优点：
简单：Celery容易使用和维护，不需要配置文件。
高可用：工作进程和客户端连接断开或者连接失败能够自动重连，并且和一些代理支持主主、主从类型的高可用。
快：单个celery进程能够每秒处理百万级的任务，使用broker和优化的配置能达到小于毫秒的延迟。
灵活：Celery的每个部分都能被扩展或者单独使用，能够客户化pool，序列化，压缩机制，日志记录，调度，消费者，生产者，传输代理等。
Celery架构
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
网友评论