在使用 Celery 4.2 版本时，遇到了内存泄漏问题，话不多说先上运行环境：
- Python 版本：2.7.5
- Celery 版本：4.2.2
- 系统：CentOS
AsyncTask 的内存泄漏
开发环境跑了一段时间之后，出现发布任务的 service 内存占用很高，而且会出现任务发布不了的情况。但发布任务的 service 并没有做什么太过复杂的操作，经过分析之后，将问题定位在了 chord 函数中会使用了 apply_async 来插入大量的 task。
复现
写了一个简单的测试脚本来测试 chord :
# -*- coding: utf-8 -*-
import resource
from celery import Celery, chord
backend_url = 'redis://redis:6379/1'
broker_url = 'pyamqp://rabbitmq:5672//?heartbeat=30'
app = Celery('task', broker=broker_url, backend=backend_url)
@app.task
def dummy():
return '1'
@app.task()
def result():
return '2'
def print_mem():
print(
'Memory usage: {} (kb)'.format(
resource.getrusage(resource.RUSAGE_SELF).ru_maxrss)
)
def run():
for i in range(10000):
chord(dummy.si())(result.si())
if i % 1000 == 0:
print_mem()
if __name__ == '__main__':
run()
每执行 1000 次打印一下内存占用情况，这个脚本在 Celery 4.2.2 版本里内存飞速增长。
Memory usage: 28041216 (kb)
Memory usage: 43773952 (kb)
Memory usage: 59592704 (kb)
Memory usage: 75333632 (kb)
Memory usage: 90955776 (kb)
Memory usage: 106758144 (kb)
Memory usage: 124145664 (kb)
Memory usage: 139677696 (kb)
Memory usage: 155234304 (kb)
Memory usage: 170778624 (kb)
分析
在使用 objgraph 进行检测之后发现，内存中驻留了大量的 promise 和 AsyncResult ，继续 debug 下去可以发现问题是在这一行 https://github.com/celery/celery/blob/v4.2.2/celery/result.py#L102 ，去掉 self.on_ready = promise(self._on_fulfilled) 后内存就能被回收了。而这个 promise 有个 weak 参数, 设置成 True 之后就会创建 self._on_fulfilled 的弱引用，AsyncResult 也能被回收，那 weak 改成 True 就对了吗?
然而并不是，搜了一下发现 Celery 以前的一个PR：Removing weak-references in promises to bound methods (Fixes #3813) #4131，当时这段代码是设置了 weak=True 的，为什么要把 weak=True 删掉呢？因为传入的 self._on_fulfilled 是 AsyncResult 对象的 bound method，weak.ref 无法处理，后续尝试获取引用的时候总会得到 None，当时的 PR 就是为了解决 self._on_fulfilled 不会被执行的问题. 要对 bound method 做弱引用需要使用 WeakMethod，但这个在 Python 3.4 版本才开始有。
好了，回到之前的问题，那么内存泄漏是哪里来的呢？
AsyncResult 中确实有循环引用， AsyncResult -> self.on_ready -> promise -> self._on_fulfilled -> self，引用计数算法对于循环引用的对象是无法回收的，而标记清除可以，但 AsyncResult 这个类还定义了 __del__ 方法，这会让 Python 的 gc 在处理循环引用的对象时不知道该以什么顺序去运行他们的 __del__ 方法, 这些对象就会一直驻留在内存里，具体可看这篇文章: gc.garbage
解决方案
- 最简单粗暴的方案是将
__del__方法删除掉，这样 Python 的 gc 就可以正确处理循环引用的对象了； - 官方解决方案：把 WeakMethod backport 到 celery 用到的异步库 vine 里去: https://github.com/celery/vine/issues/21 ，这样在 celery 那头设置
weak=True就能正确处理了。简单来说就是将 Celery 升级到 4.3.0 版本。
参考链接