celery 源码阅读1

本文主要是介绍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的实现，并不需要显示的编写继承。

这篇关于celery 源码阅读1的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！

C/C++教程