celery源码阅读1

celery的项目结构

celery我们选用 5.0.5 版本，首先requirements/default.txt文件描述主要依赖下面几个库:

• billiard（台球） celery项目提供的一个多进程池的实现


• kombu（命令） celery项目提供的一个消息库，可以对接不同的消息队列，比如RabbitMQ，Redis


• vine（葡萄藤） celery项目提供的一个promise实现，可以处理任务的组合和pipline等

celery任务执行的结果也支持多种存储方式:

• Mongo


• Redis


• Elasticsearch


• ...

celery的并发也支持多种实现

• 多进程的fork


• gevent


• 多线程


• eventlet


• ...

celery支持工作流

• 可以根据函数签名进行调度


• 可以支持链式任务


• 可以支持分组，和弦...

promise库的实现

promise在异步任务中非常重要，所以celery有个vine项目实现了promise功能，在开始celery之前，我们先扫清这些外围障碍。

promise 简介

Promise是一个对象，它代表了一个异步操作的最终完成或者失败。我觉得MDN中的介绍非常好，我们先了解它，再对比看看Python中如何实现它。比如有一个创建音频文件的操作，成功和失败的时候使用不同的输出:

// 成功的回调函数
function successCallback(result) {
console.log("音频文件创建成功: " + result);
}
// 失败的回调函数
function failureCallback(error) {
console.log("音频文件创建失败: " + error);
}
createAudioFileAsync(audioSettings, successCallback, failureCallback)

传统的方式就是使用callback方式调用，把正确和错误的回调传入执行函数createAudioFileAsync中。如果使用Promise方式就会变成:

# 创建一个Promise对象
const promise = createAudioFileAsync(audioSettings);
# 执行这个Promise对象
promise.then(successCallback, failureCallback);

doSomething().then(function(result) {
return doSomethingElse(result);
})
.then(function(newResult) {
return doThirdThing(newResult);
})
.then(function(finalResult) {
console.log('Got the final result: ' + finalResult);
})
.catch(failureCallback);

可以发现使用Promise方式后，代码会变扁平，非常清爽。这里展示了Promise的2个特点:

• then函数执行后返回的是一个Promise对象


• Promise可以进行链式(chain)调用

简单理解Promise后，我们再回头看doSomething，doSomethingElse和doThirdThing，如果这是3个任务，需要按顺序调度执行，后者需要前者的执行结果作为参数？所以可以知道，从逻辑上讲，Promise功能对于任务调度，非常重要。这种实现方式，和语言其实无关。

Promise的实现

先单元测试用例看从promise的使用:

def test_signal(self):
# mock一个函数
callback = Mock(name='callback')
# 创建一个Promise对象
a = promise()
# 使用then函数添加callback
a.then(callback)
# 执行Promise对象
a(42)
# 函数被调用，参数为42
callback.assert_called_once_with(42)

 不难看出python版本的Promise使用和Javascript版本没有太大区别。promise的构造函数如下:

class promise:
...
def __init__(self, fun=None, args=None, kwargs=None,
callback=None, on_error=None, weak=False,
ignore_result=False):
self.weak = weak
self.ignore_result = ignore_result
# 要执行的函数
self.fun = self._get_fun_or_weakref(fun=fun, weak=weak)
# 注意位置参数是元祖，这样才可以在call里叠加
self.args = args or ()
# 关键字参数是字典
self.kwargs = kwargs or {}
# ready，failed，cancelled 三个状态，默认都是false
self.ready = False
self.failed = False
self.value = None
self.reason = None
# Optimization
# Most promises will only have one callback, so we optimize for this
# case by using a list only when there are multiple callbacks.
# s(calar) pending / l(ist) pending
# 单个callback/多个callback
self._svpending = None
self._lvpending = None
self.on_error = on_error
self.cancelled = False
# 可见callback可以通过参数传递，也可以通过then函数传递
if callback is not None:
self.then(callback)

每个Promise有3个状态位:ready，cancelled和failed，默认都是false。

重点就是then函数:

def then(self, callback, on_error=None):
# callback是普通函数，就用promise再嵌套一下
if not isinstance(callback, Thenable):
callback = promise(callback, on_error=on_error)
if self.cancelled:
callback.cancel()
return callback
if self.failed:
callback.throw(self.reason)
elif self.ready:
args, kwargs = self.value
callback(*args, **kwargs)
if self._lvpending is None:
svpending = self._svpending
if svpending is not None:
self._svpending, self._lvpending = None, deque([svpending])
else:
# 初始复制callback给_svpending
# 就是一种递归
self._svpending = callback
return callback
# 添加到右侧
self._lvpending.append(callback)
# 返回的是一个promise可以继续then，实现a.then(fun_x).then(fun_y).then(fun_z)这样的链式调用
return callback

在Javascript版本已经介绍过Promise的2个特性就是，then返回一个新的Promise对象，又由于返回的是Promise对象，又可以继续执行then函数添加新的callback链式调用。

Promise对象的执行就在魔法函数 *__call__:

def __call__(self, *args, **kwargs):
retval = None
if self.cancelled:
return
# 叠加参数
final_args = self.args + args if args else self.args
final_kwargs = dict(self.kwargs, **kwargs) if kwargs else self.kwargs
# self.fun may be a weakref
fun = self._fun_is_alive(self.fun)
if fun is not None:
try:
if self.ignore_result:
fun(*final_args, **final_kwargs)
ca = ()
ck = {}
else:
# 执行函数
retval = fun(*final_args, **final_kwargs)
self.value = (ca, ck) = (retval,), {}
except Exception:
# 异常
return self.throw()
else:
self.value = (ca, ck) = final_args, final_kwargs
# 更改ready状态
self.ready = True
svpending = self._svpending
# 执行callback，把fun执行的结果往callback里传入这样形成pipeline
if svpending is not None:
try:
svpending(*ca, **ck)
finally:
self._svpending = None
else:
lvpending = self._lvpending
try:
while lvpending:
# 从左开始执行
p = lvpending.popleft()
p(*ca, **ck)
finally:
self._lvpending = None
return retval

上述代码主要步骤有：

• 合并Promise对象的参数和函数调用参数


• 如果有初始函数，则执行初始函数


• 更改Promise对象的ready状态


• 执行callback

链式调用的示例，请看:

def test_chained(self):
def add(x, y):
return x + y
def pow2(x):
return x ** 2
adder = Mock(name='adder')
adder.side_effect = add
power = Mock(name='multiplier')
power.side_effect = pow2
final = Mock(name='final')
p = promise()
# 链式调用(注意是有序的)
p.then(adder).then(power).then(final)
p(42, 42)
assert p.value == ((42, 42), {})
adder.assert_called_with(42, 42)
power.assert_called_with(84)
final.assert_called_with(7056)

Barrier的实现

vine还提供了一个叫做barrier的实现，处理多个Promise对象的串行化，下面是单元测试:

class test_barrier:
def setup(self):
self.m1, self.m2, self.m3 = Mock(), Mock(), Mock()
self.ps = [promise(self.m1), promise(self.m2), promise(self.m3)]
def test_evaluate(self):
# 需要执行4才才变成ready
x = barrier(self.ps)
x()
assert not x.ready
x()
assert not x.ready
x.add(promise())
x()
assert not x.ready
x()
assert x.ready
x()
x()
# 已经执行完成继续添加会报错
with pytest.raises(ValueError):
x.add(promise())

使用barrier后，4个Promise对象的调用可以串行化，并且可以单步执行。执行一次x()消耗一个Promise。

barrier的构造函数:

class barrier:

def __init__(self, promises=None, args=None, kwargs=None,
callback=None, size=None):
# Promise的实现
self.p = promise()
self.args = args or ()
self.kwargs = kwargs or {}
self._value = 0
self.size = size or 0
if not self.size and promises:
# iter(l) calls len(l) so generator wrappers
# can only return NotImplemented in the case the
# generator is not fully consumed yet.
plen = promises.__len__()
if plen is not NotImplemented:
self.size = plen
self.ready = self.failed = False
self.reason = None
self.cancelled = False
self.finalized = False
# 列表推导式
[self.add_noincr(p) for p in promises or []]
self.finalized = bool(promises or self.size)
if callback:
self.then(callback)

barrier重点是默认有一个promise实现，用来作为整个批处理的尾部。参数中的promises列表通过add_noincr函数形成调用链:

def add_noincr(self, p):
if not self.cancelled:
# 已经完成了就不能够再添加了
if self.ready:
raise ValueError('Cannot add promise to full barrier')
# 其实就是then().then().then() 添加到自己之前，自己主要执行最开始定义的callback
p.then(self)

每执行一次进行计数，直到执行完成后更改状态和执行自身(尾部)的Promise对象

def __call__(self, *args, **kwargs):
# 判断是否已经执行完成：ready和cancelled
if not self.ready and not self.cancelled:
self._value += 1
if self.finalized and self._value >= self.size:
self.ready = True
self.p(*self.args, **self.kwargs)

对于抽象的实现，在Python中除了可以使用继承方式，还可以使用组合Mixin。比如下面：

class Thenable(Callable, metaclass=abc.ABCMeta): # pragma: no cover
...
@abc.abstractmethod
def then(self, on_success, on_error=None):
"""成功和失败的2个回调"""
raise NotImplementedError()

class CanThen:
def then(self, x, y):
pass

assert isinstance(CanThen(), Thenable)

可以看到CanThen实现then函数后，就可以被认定为Thenable的实现，但是CanThen并未继承自Thenable。这个魔法主要是由ABCMeta的register和__subclasshook__两个方法实现:

class Thenable(Callable, metaclass=abc.ABCMeta):

...

@classmethod
def __subclasshook__(cls, C):
# 也由ABCMeta提供
if cls is Thenable:
if any('then' in B.__dict__ for B in C.__mro__):
return True
return NotImplemented
@classmethod
def register(cls, other):
# overide to return other so `register` can be used as a decorator
# 这个register方法是由ABCMeta提供，其实现类使用装饰器方式
# https://docs.python.org/zh-cn/3/library/abc.html
type(cls).register(cls, other)
return other
@Thenable.register
class promise:
pass
assert isinstance(promise(lambda x: x), Thenable)

promise类经过Thenable.register类装饰圈注释一下后，就可以被认定位Thenable的实现，并不需要显示的编写继承。