celery最佳实践

做为一个Celery使用重度用户。看到Celery Best Practices这篇文章。不禁得菊花一紧。

干脆翻译出来，同一时候也会添加咱们项目中celery的实战经验。

至于Celery为什么物，看这里Celery。

一般在使用Django的时候，你可能需要运行一些长时间的后台任务，没准你可能需要使用一些能排序的任务队列，那么Celery将会是一个很是好的选择。

当把Celery做为一个任务队列用于很是多项目中后，做者积累了一些最佳实践方式，譬如怎样用合适的方式使用Celery，以及一些Celery提供的但是还未充分使用的特性。

1，不要使用数据库做为你的AMQP Broker

数据库并不是天生设计成能用于AMQP broker的。在生产环境下，它很是有可能在某时候当机（PS，当掉这点我认为不论什么系统都不能保证不当吧！！

。）。

做者猜测为啥很是多人使用数据库做为broker主要是因为他们已经有一个数据库用来给web app提供数据存储了。因而干脆直接拿来使用。设置成Celery的broker是很是easy的。并且不需要再安装其它组件（譬如RabbitMQ）。

假设有例如如下场景：你有4个后端workers去获取并处理放入到数据库里面的任务，这意味着你有4个进程为了获取最新任务，需要频繁地去轮询数据库。没准每个worker同一时候还有多个本身的并发线程在干这事情。

某一天。你发现因为太多的任务产生。4个worker不够用了，处理任务的速度已经大大落后于生产任务的速度，因而你不停去添加worker的数量。忽然，你的数据库因为大量进程轮询任务而变得响应缓慢，磁盘IO一直处于高峰值状态，你的web应用也開始受到影响。这一切，都因为workers在不停地对数据库进行DDOS。

而当你使用一个合适的AMQP（譬如RabbitMQ）的时候，这一切都不会发生，以RabbitMQ为例。首先，它将任务队列放到内存里面。你不需要去訪问硬盘。其次，consumers（也就是上面的worker）并不需要频繁地去轮询因为RabbitMQ能将新的任务推送给consumers。

固然，假设RabbitMQ真出现故障了。至少也不会影响到你的web应用。

这也就是做者说的不用数据库做为broker的缘由，并且很是多地方都提供了编译好的RabbitMQ镜像，你都能直接使用，譬如这些。

对于这点。我是深表赞同的。咱们系统大量使用Celery处理异步任务，大概平均一天几百万的异步任务，曾经咱们使用的mysql。而后总会出现任务处理延时太严重的问题，即便添加了worker也无论用。因而咱们使用了redis。性能提高了很是多。至于为啥使用mysql很是慢，咱们没去深究，没准也还真出现了DDOS的问题。

2，使用不少其它的queue（不要仅仅用默认的）

Celery很是easy设置，一般它会使用默认的queue用来存听任务（除非你显示指定其它queue）。一般写法例如如下：

@app.task()
def my_taskA(a, b, c):
    print("doing something here...")

@app.task()
def my_taskB(x, y):
    print("doing something here...")

这两个任务都会在同一个queue里面运行。这样写事实上很是有吸引力的，因为你仅仅需要使用一个decorator就能实现一个异步任务。做者关心的是taskA和taskB没准是全然两个不一样的东西，或者一个可能比还有一个更加剧要，那么为何要把它们放到一个篮子里面呢？（鸡蛋都不能放到一个篮子里面，是吧！）没准taskB事实上不怎么重要，但是量太多，以致于重要的taskA反而不能高速地被worker进行处理。添加workers也解决不了这个问题，因为taskA和taskB仍然在一个queue里面运行。

3。使用具备优先级的workers

为了解决2里面出现的问题，咱们需要让taskA在一个队列Q1，而taskB在还有一个队列Q2运行。同一时候指定x workers去处理队列Q1的任务，而后使用其它的workers去处理队列Q2的任务。使用这样的方式，taskB能够得到足够的workers去处理，同一时候一些优先级workers也能很是好地处理taskA而不需要进行长时间的等待。

首先手动定义queue

CELERY_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('for_task_A', Exchange('for_task_A'), routing_key='for_task_A'),
    Queue('for_task_B', Exchange('for_task_B'), routing_key='for_task_B'),
)

而后定义routes用来决定不一样的任务去哪个queue

CELERY_ROUTES = {
    'my_taskA': {'queue': 'for_task_A', 'routing_key': 'for_task_A'},
    'my_taskB': {'queue': 'for_task_B', 'routing_key': 'for_task_B'},
}

最后再为每个task启动不一样的workerscelery worker -E -l INFO -n workerA -Q for_task_A celery worker -E -l INFO -n workerB -Q for_task_B

在咱们项目中。会涉及到大量文件转换问题，有大量小于1mb的文件转换，同一时候也有少许将近20mb的文件转换。小文件转换的优先级是最高的，同一时候不用占用很是多时间，但大文件的转换很是耗时。假设将转换任务放到一个队列里面，那么很是有可能因为出现转换大文件，致使耗时太严重形成小文件转换延时的问题。

因此咱们依照文件大小设置了3个优先队列。并且每个队列设置了不一样的workers。很是好地攻克了咱们文件转换的问题。

4，使用Celery的错误处理机制

大多数任务并无使用错误处理，假设任务失败，那就失败了。在一些状况下这很是不错。但是做者见到的多数失败任务都是去调用第三方API而后出现了网络错误。或者资源不可用这些错误。而对于这些错误。最简单的方式就是重试一下，或许就是第三方API暂时服务或者网络出现故障，没准当即就行了，那么为何不试着重试一下呢？

@app.task(bind=True, default_retry_delay=300, max_retries=5)
def my_task_A():
    try:
        print("doing stuff here...")
    except SomeNetworkException as e:
        print("maybe do some clenup here....")
        self.retry(e)

做者喜欢给每个任务定义一个等待多久重试的时间，以及最大的重试次数。固然还有更具体的參数设置，本身看文档去。

对于错误处理，咱们因为使用场景特殊，好比一个文件转换失败，那么无论多少次重试都会失败。因此没有添加剧试机制。

5，使用Flower

Flower是一个很是强大的工具，用来监控celery的tasks和works。

这玩意咱们也没怎么使用。因为多数时候咱们都是直接链接redis去查看celery相关状况了。貌似挺傻逼的对不，尤为是celery在redis里面存放的数据并不能方便的取出来。

6，没事别太关注任务退出状态

一个任务状态就是该任务结束的时候成功仍是失败信息，没准在一些统计场合，这很是实用。但咱们需要知道。任务退出的状态并不是该任务运行的结果，该任务运行的一些结果因为会对程序有影响，通常会被写入数据库（好比更新一个用户的朋友列表）。

做者见过的多数项目都将任务结束的状态存放到sqlite或者本身的数据库，但是存这些真有必要吗，没准可能影响到你的web服务的。因此做者一般设置CELERY_IGNORE_RESULT = True去丢弃。

对于咱们来讲，因为是异步任务，知道任务运行完毕以后的状态真没啥用。因此果断丢弃。

7，不要给任务传递 Database/ORM 对象

这个事实上就是不要传递Database对象（好比一个用户的实例）给任务。因为没准序列化以后的数据已是过时的数据了。

因此不妨直接传递一个user id，而后在任务运行的时候实时的从数据库获取。

对于这个，咱们也是如此，给任务仅仅传递相关id数据。譬如文件转换的时候，咱们仅仅会传递文件的id，而其它文件信息的获取咱们都是直接经过该id从数据库里面取得。

最后

后面就是咱们本身的感触了，上面做者提到的Celery的使用，真的能够算是很是好地实践方式。至少现在咱们的Celery没出过太大的问题，固然小坑仍是有的。至于RabbitMQ，这玩意咱们是真没用过，效果怎么样不知道。至少比mysql好用吧。

最后。附上做者的一个Celery Talk https://denibertovic.com/talks/celery-best-practices/。