一个完整的Tasks生命周期

这个时候我们终于可以解答我们在Celery基础架构文末留出来的第二个问题了：

我们定义的task celery 是如何扫描并注册到celery的？

我们知道，在python中，貌似并没有java哪种注解的概念，也没有发现类似于注解扫描的机制，那celery是如何知道我们哪些函数是被@task修饰过并添加到上下文的呢？

在详细分析源码之前，我们需要先搞清楚一个概念，还记得我们在之前的文章中曾分析了task的协议，其中task_id这些源信息是放在header里面的，参数是放在body里面的，这说明了一个非常有趣的事情，celery貌似并没有把函数作为一个单元作为消息体在消息队列中传输，而是只告诉了worker对应的任务的名称以及参数。为什么不把函数作为一整个计算单元传到worker呢，worker拿到代码直接执行不就好了？

celery这样做的好处非常的明显：那就是极大的压缩了消息的体积，因为一整个函数传输的数据量通常来说并不小。这波叫节省数据传输的成本，提高效率。

但缺点也是显而易见的，那就是worker和我们代码形成了比较深的耦合，即worker必须要从我们项目启动，不然无法根据task_name 去找到并执行相关的task的逻辑。

既然我们通常声明一个celery 任务的方式就是给对应的函数加上@task的这个装饰器，那不妨我们就从这里下手。

@Task 装饰器

def task(*args, **kwargs):
    """Deprecated decorator, please use :func:`celery.task`."""
    return current_app.task(*args, **dict({'base': Task}, **kwargs))

点进去一看，非常的简单的，注意，这里的current_app是我们Worker中全局的Celery对象的实例。这个装饰器的实际作用就是把我们的函数包装成了一个Task的代理对象，不然你的函数哪里来的delay和apply_ansyc方法，是吧，非常有道理。而这里的task方法，注意，就是我们前面说非常非常重要的Celery类的task方法。再一次把代码贴上来。对应的代码路径: celery.app.base.Celery

def task(self, *args, **opts):

    if USING_EXECV and opts.get('lazy', True):
        # When using execv the task in the original module will point to a
        # different app, so doing things like 'add.request' will point to
        # a different task instance.  This makes sure it will always use
        # the task instance from the current app.
        # Really need a better solution for this :(
        from . import shared_task
        return shared_task(*args, lazy=False, **opts)

    def inner_create_task_cls(shared=True, filter=None, lazy=True, **opts):
        _filt = filter

        def _create_task_cls(fun):
            if shared:
                def cons(app):
                    return app._task_from_fun(fun, **opts)
                cons.__name__ = fun.__name__
                connect_on_app_finalize(cons)
            if not lazy or self.finalized:
                ret = self._task_from_fun(fun, **opts)
            else:
              	# 生成一个代理对象
                # return a proxy object that evaluates on first use
                ret = PromiseProxy(self._task_from_fun, (fun,), opts,
                                    __doc__=fun.__doc__)
                # 放到待处理的队列里面，注意，这部分和connect_on_app_finalize(cons)会各执行一次。
                self._pending.append(ret)
            if _filt:
                return _filt(ret)
            return ret

        return _create_task_cls

    if len(args) == 1:
        if callable(args[0]):
            return inner_create_task_cls(**opts)(*args)
        raise TypeError('argument 1 to @task() must be a callable')
    if args:
        raise TypeError(
            '@task() takes exactly 1 argument ({0} given)'.format(
                sum([len(args), len(opts)])))
    return inner_create_task_cls(**opts)

注意inner_create_task_cls, 最终我们把task包装成了一个这玩意。首先我们需要特别留意两个地方，一个就是connect_on_app_finalize。因为shared默认值是True，所以一定会执行这行代码。点进去看看：

_on_app_finalizers = set()

def connect_on_app_finalize(callback):
    """Connect callback to be called when any app is finalized."""
    _on_app_finalizers.add(callback)
    return callback

到这里我们发现一个惊天小秘密，在celery内部，维护了一个set，这个set的作用现在我们也不知道是啥。但是我们知道他把一个函数给add进去了，cons这个函数。他调用了_task_from_fun方法。点进去看看:

def _task_from_fun(self, fun, name=None, base=None, bind=False, **options):
    if not self.finalized and not self.autofinalize:
        raise RuntimeError('Contract breach: app not finalized')
    name = name or self.gen_task_name(fun.__name__, fun.__module__)
    base = base or self.Task
    if name not in self._tasks:
        run = fun if bind else staticmethod(fun)
        task = type(fun.__name__, (base,), dict({
            'app': self,
            'name': name,
            'run': run,
            '_decorated': True,
            '__doc__': fun.__doc__,
            '__module__': fun.__module__,
            '__header__': staticmethod(head_from_fun(fun, bound=bind)),
            '__wrapped__': run}, **options))()
        # for some reason __qualname__ cannot be set in type()
        # so we have to set it here.
        try:
            task.__qualname__ = fun.__qualname__
        except AttributeError:
            pass
        # 注册到app维护的列表里面
        self._tasks[task.name] = task
        # 这里的作用是吧app的各种属性复制到task内部去
        task.bind(self)  # connects task to this app
        """
        ...... 省略一部分不重要的
        """
        if autoretry_for and not hasattr(task, '_orig_run'):

            @wraps(task.run)
            def run(*args, **kwargs):
                try:
                    return task._orig_run(*args, **kwargs)
                except Ignore:
                    # If Ignore signal occures task shouldn't be retried,
                    # even if it suits autoretry_for list
                    raise
                except Retry:
                    raise
                except autoretry_for as exc:
                    if retry_backoff:
                        retry_kwargs['countdown'] = \
                            get_exponential_backoff_interval(
                                factor=retry_backoff,
                                retries=task.request.retries,
                                maximum=retry_backoff_max,
                                full_jitter=retry_jitter)
                    raise task.retry(exc=exc, **retry_kwargs)

            task._orig_run, task.run = task.run, run
    else:
        task = self._tasks[name]
    return task

这段代码的可读性出神入化了已经，name的话比较容易理解, 如果你指定了就用你指定的，如果你没指定，那就自动给你生成一个，生成的规则是:itsm.ticket.tasks.notify_task其实就是路径。这里的fun其实就是我们的函数对象。

我们看到首先是用type关键字给生成一个task对象，这里的task对象是Task类的实例。然后把它加到了app的_tasks列表里面。然后将task与app进行绑定的。特别要注意的是self._tasks这个属性，里面的内容大致上长这个样子:

{
  "itsm.ticket.tasks.notify_task": <@task: itsm.ticket.tasks.notify_task of proj at 0x7fbcecd3d9b0 (v2 compatible)>
}

这样一来我们的task_name 和对应的 执行对象就有了

注意哦，我们的_on_app_finalizers存的可不是什么函数实例，当时我们cons这个函数传进去了，也就是函数此刻还是没执行的。那究竟是在哪里执行的呢？ 这个需要我们回到Worker的初始化过程中去。我们前面有提到，Worker对象在初始化过程中有执行到on_before_init函数，然后在这个函数里面我们发现了一段代码: trace.setup_worker_optimizations(self.app, self.hostname)

class Worker(WorkController):
    """Worker as a program."""

    def on_before_init(self, quiet=False, **kwargs):
        self.quiet = quiet
        # 就是这一行
        trace.setup_worker_optimizations(self.app, self.hostname)

        # this signal can be used to set up configuration for
        # workers by name.
        signals.celeryd_init.send(
            sender=self.hostname, instance=self,
            conf=self.app.conf, options=kwargs,
        )
        check_privileges(self.app.conf.accept_content)

当我们赶到setup_worker_optimizations现场的时候，就只发现了这段代码:

# evaluate all task classes by finalizing the app.
app.finalize()

再次回到Celery这个类的finalize方法。

def finalize(self, auto=False):
    """Finalize the app.

    This loads built-in tasks, evaluates pending task decorators,
    reads configuration, etc.
    """
    with self._finalize_mutex:
        if not self.finalized:
            if auto and not self.autofinalize:
                raise RuntimeError('Contract breach: app not finalized')
            self.finalized = True
            _announce_app_finalized(self)

            pending = self._pending
            while pending:
                maybe_evaluate(pending.popleft())

            for task in values(self._tasks):
                task.bind(self)

            self.on_after_finalize.send(sender=self)

发现这样一行代码，点进去看看_announce_app_finalized(self):

def _announce_app_finalized(app):
    callbacks = set(_on_app_finalizers)
    for callback in callbacks:
        callback(app)

执行到这一步，发现，诶，我们的self._tasks 终于算是完整了。这里的设计非常的绕，可能需要多看看才能搞清楚这之间的调用关系。

到这里我们几乎已经回答了上面提到的那个问题，Task是如何被注册到全局上下文的。当我么import某个模块的时候，装饰器会自动执行，所以动态加载项目下所有模块的时候，task就会被注册到celery里面去了。

任务的消费

讲完任务如何发现了，那下面就需要去梳理任务是如何被消费的。在celery中维护了多进程的任务消费模型，但是任务执行的细节并不在本文的范畴中，可能会放到后面的章节中去梳理。

大家现在停下来仔细的思考一下，如果说任务的注册是从@task作为入口开始的，那任务消费的入口又是哪里？

那必然是从接收到消息的那一刻开始的。因为只有接受到了任务才可以开启任务的消费过程。还记得我们在Consumer中发现的那个有用的函数么，on_task_received, 它和kombu的消息回调函数绑定到了一块，也就是当监听到消息队列中有新消息的时候，就会调用on_task_received。

def create_task_handler(self, promise=promise):
    strategies = self.strategies
    on_unknown_message = self.on_unknown_message
    on_unknown_task = self.on_unknown_task
    on_invalid_task = self.on_invalid_task
    callbacks = self.on_task_message
    call_soon = self.call_soon

    def on_task_received(message):
        # payload will only be set for v1 protocol, since v2
        # will defer deserializing the message body to the pool.
        payload = None
        try:
            type_ = message.headers['task']                # protocol v2
        except TypeError:
            return on_unknown_message(None, message)
        except KeyError:
            try:
                payload = message.decode()
            except Exception as exc:  # pylint: disable=broad-except
                return self.on_decode_error(message, exc)
            try:
                type_, payload = payload['task'], payload  # protocol v1
            except (TypeError, KeyError):
                return on_unknown_message(payload, message)
        try:
            strategy = strategies[type_]
        except KeyError as exc:
            return on_unknown_task(None, message, exc)
        else:
            try:
                strategy(
                    message, payload,
                    promise(call_soon, (message.ack_log_error,)),
                    promise(call_soon, (message.reject_log_error,)),
                    callbacks,
                )
            except (InvalidTaskError, ContentDisallowed) as exc:
                return on_invalid_task(payload, message, exc)
            except DecodeError as exc:
                return self.on_decode_error(message, exc)

    return on_task_received

这里有个地方非常值得我们注意，self.strategies, 我们的tasks呢？？？，上面讲半天，好家伙，根本没用着。

大可不必，其中必有引擎，我们找一下self.strategies的初始化逻辑:

def update_strategies(self):
    loader = self.app.loader
    for name, task in items(self.app.tasks):
        self.strategies[name] = task.start_strategy(self.app, self)
        task.__trace__ = build_tracer(name, task, loader, self.hostname,
                                      app=self.app)

这不联系上了吗，再点进去start_sreategy方法看看是干啥的:

def start_strategy(self, app, consumer, **kwargs):
    return instantiate(self.Strategy, self, app, consumer, **kwargs)

发现是生成了一个Strategy对象的实例。追查发现，Strategy对应的位置是: celery.worker.strategy:default,点进去看看，说不定能发现什么意外惊喜呢。

def default(task, app, consumer,
            info=logger.info, error=logger.error, task_reserved=task_reserved,
            to_system_tz=timezone.to_system, bytes=bytes, buffer_t=buffer_t,
            proto1_to_proto2=proto1_to_proto2):
    """Default task execution strategy.

    Note:
        Strategies are here as an optimization, so sadly
        it's not very easy to override.
    """
    hostname = consumer.hostname
    connection_errors = consumer.connection_errors
    _does_info = logger.isEnabledFor(logging.INFO)

    # task event related
    # (optimized to avoid calling request.send_event)
    eventer = consumer.event_dispatcher
    events = eventer and eventer.enabled
    send_event = eventer and eventer.send
    task_sends_events = events and task.send_events

    call_at = consumer.timer.call_at
    apply_eta_task = consumer.apply_eta_task
    rate_limits_enabled = not consumer.disable_rate_limits
    get_bucket = consumer.task_buckets.__getitem__
    handle = consumer.on_task_request
    limit_task = consumer._limit_task
    limit_post_eta = consumer._limit_post_eta
    body_can_be_buffer = consumer.pool.body_can_be_buffer
    Request = symbol_by_name(task.Request)
    Req = create_request_cls(Request, task, consumer.pool, hostname, eventer)

    revoked_tasks = consumer.controller.state.revoked

    def task_message_handler(message, body, ack, reject, callbacks,
                             to_timestamp=to_timestamp):
        if body is None and 'args' not in message.payload:
            body, headers, decoded, utc = (
                message.body, message.headers, False, app.uses_utc_timezone(),
            )
            if not body_can_be_buffer:
                body = bytes(body) if isinstance(body, buffer_t) else body
        else:
            if 'args' in message.payload:
                body, headers, decoded, utc = hybrid_to_proto2(message,
                                                               message.payload)
            else:
                body, headers, decoded, utc = proto1_to_proto2(message, body)

        req = Req(
            message,
            on_ack=ack, on_reject=reject, app=app, hostname=hostname,
            eventer=eventer, task=task, connection_errors=connection_errors,
            body=body, headers=headers, decoded=decoded, utc=utc,
        )
        if _does_info:
            info('Received task: %s', req)
        if (req.expires or req.id in revoked_tasks) and req.revoked():
            return

        signals.task_received.send(sender=consumer, request=req)

        if task_sends_events:
            send_event(
                'task-received',
                uuid=req.id, name=req.name,
                args=req.argsrepr, kwargs=req.kwargsrepr,
                root_id=req.root_id, parent_id=req.parent_id,
                retries=req.request_dict.get('retries', 0),
                eta=req.eta and req.eta.isoformat(),
                expires=req.expires and req.expires.isoformat(),
            )

        bucket = None
        eta = None
        if req.eta:
            try:
                if req.utc:
                    eta = to_timestamp(to_system_tz(req.eta))
                else:
                    eta = to_timestamp(req.eta, app.timezone)
            except (OverflowError, ValueError) as exc:
                error("Couldn't convert ETA %r to timestamp: %r. Task: %r",
                      req.eta, exc, req.info(safe=True), exc_info=True)
                req.reject(requeue=False)
        if rate_limits_enabled:
            bucket = get_bucket(task.name)

        if eta and bucket:
            consumer.qos.increment_eventually()
            return call_at(eta, limit_post_eta, (req, bucket, 1),
                           priority=6)
        if eta:
            consumer.qos.increment_eventually()
            call_at(eta, apply_eta_task, (req,), priority=6)
            return task_message_handler
        if bucket:
            return limit_task(req, bucket, 1)

        task_reserved(req)
        if callbacks:
            [callback(req) for callback in callbacks]
        handle(req)
    return task_message_handler

哇靠，有点意思，也就是说，celery再收到消息之后，马上调用了这个消息对应的执行策略，默认的是default, 点进去看看它到底干了啥。注意，实际上strategies对应的value是task_message_handler这个函数。

这里特别需要留意的地方是这一句。

Req = create_request_cls(Request, task, consumer.pool, hostname, eventer)

点进去看看创建了个啥，嘿，有点意思:

def create_request_cls(base, task, pool, hostname, eventer,
                       ref=ref, revoked_tasks=revoked_tasks,
                       task_ready=task_ready, trace=trace_task_ret):
    default_time_limit = task.time_limit
    default_soft_time_limit = task.soft_time_limit
    apply_async = pool.apply_async
    acks_late = task.acks_late
    events = eventer and eventer.enabled

    class Request(base):

        def execute_using_pool(self, pool, **kwargs):
            task_id = self.task_id
            if (self.expires or task_id in revoked_tasks) and self.revoked():
                raise TaskRevokedError(task_id)

            time_limit, soft_time_limit = self.time_limits
            # 注意这里
            result = apply_async(
                trace,
                args=(self.type, task_id, self.request_dict, self.body,
                      self.content_type, self.content_encoding),
                accept_callback=self.on_accepted,
                timeout_callback=self.on_timeout,
                callback=self.on_success,
                error_callback=self.on_failure,
                soft_timeout=soft_time_limit or default_soft_time_limit,
                timeout=time_limit or default_time_limit,
                correlation_id=task_id,
            )
            # cannot create weakref to None
            # pylint: disable=attribute-defined-outside-init
            self._apply_result = maybe(ref, result)
            return result

        def on_success(self, failed__retval__runtime, **kwargs):
            failed, retval, runtime = failed__retval__runtime
            if failed:
                if isinstance(retval.exception, (
                        SystemExit, KeyboardInterrupt)):
                    raise retval.exception
                return self.on_failure(retval, return_ok=True)
            task_ready(self)

            if acks_late:
                self.acknowledge()

            if events:
                self.send_event(
                    'task-succeeded', result=retval, runtime=runtime,
                )

    return Request

这里实际上返回了一个内置类，Request, 他有一个方法 execute_using_pool,然后它还调用了apply_async, 注意，这里真的就开始执行我们的任务了，因为任务名，参数啥的都给传进去了。 第二个从方法名上也可以看到，大概率是和并发有关系的，不过这个不属于本章的重点，让我们现在回到task_message_handler函数去，这次我们只看重点代码:

def task_message_handler(message, body, ack, reject, callbacks,
                         to_timestamp=to_timestamp):

    req = Req(
        message,
        on_ack=ack, on_reject=reject, app=app, hostname=hostname,
        eventer=eventer, task=task, connection_errors=connection_errors,
        body=body, headers=headers, decoded=decoded, utc=utc,
    )
    # 打印日志
    if _does_info:
        info('Received task: %s', req)
    if (req.expires or req.id in revoked_tasks) and req.revoked():
        return
	  # 发送信号
    signals.task_received.send(sender=consumer, request=req)
    # 发送事件
    if task_sends_events:
        send_event(
            'task-received',
            uuid=req.id, name=req.name,
            args=req.argsrepr, kwargs=req.kwargsrepr,
            root_id=req.root_id, parent_id=req.parent_id,
            retries=req.request_dict.get('retries', 0),
            eta=req.eta and req.eta.isoformat(),
            expires=req.expires and req.expires.isoformat(),
        )

    bucket = None
    eta = None
    # 处理延时任务
    if req.eta:
        try:
            if req.utc:
                eta = to_timestamp(to_system_tz(req.eta))
            else:
                eta = to_timestamp(req.eta, app.timezone)
        except (OverflowError, ValueError) as exc:
            error("Couldn't convert ETA %r to timestamp: %r. Task: %r",
                  req.eta, exc, req.info(safe=True), exc_info=True)
            req.reject(requeue=False)
    if rate_limits_enabled:
        bucket = get_bucket(task.name)

    if eta and bucket:
        consumer.qos.increment_eventually()
        return call_at(eta, limit_post_eta, (req, bucket, 1),
                       priority=6)
    if eta:
        consumer.qos.increment_eventually()
        call_at(eta, apply_eta_task, (req,), priority=6)
        return task_message_handler
    if bucket:
        return limit_task(req, bucket, 1)

    task_reserved(req)
    # 如果注册了回调函数，则调用回调函数
    if callbacks:
        [callback(req) for callback in callbacks]
    # 重点来了，重点来了
    handle(req)
return task_message_handler

通过查看源码，handle实际上绑定的函数是handle = consumer.on_task_request，consumer.on_task_request也不是Consumer类的，是初始化的时候传进去的，继续往上找:

class Consumer(bootsteps.StartStopStep):
    """Bootstep starting the Consumer blueprint."""

    last = True

    def create(self, w):
        if w.max_concurrency:
            prefetch_count = max(w.max_concurrency, 1) * w.prefetch_multiplier
        else:
            prefetch_count = w.concurrency * w.prefetch_multiplier
        c = w.consumer = self.instantiate(
            w.consumer_cls, w.process_task,
            hostname=w.hostname,
            task_events=w.task_events,
            init_callback=w.ready_callback,
            initial_prefetch_count=prefetch_count,
            pool=w.pool,
            timer=w.timer,
            app=w.app,
            controller=w,
            hub=w.hub,
            worker_options=w.options,
            disable_rate_limits=w.disable_rate_limits,
            prefetch_multiplier=w.prefetch_multiplier,
        )
        return c

发现consumer.on_task_request实际指向的是 w.process_task, 有点眉目了，继续往Worker找，只发现了这个:

def _process_task(self, req):
    """Process task by sending it to the pool of workers."""
    try:
        req.execute_using_pool(self.pool)
    except TaskRevokedError:
        try:
            self._quick_release()   # Issue 877
        except AttributeError:
            pass

只到这里，才把execute_using_pool调用了。但是问题出现了，_process_task它多了个下划线啊，在Worker类中，我们并没有发现process_task这个函数，根据我们的经验，只能懵了，还记得我说execute_using_pool 看起来和多进程有关系吗？ 就算没有关系，那大概率也和Pool有点关系，于是我们找Worker的子组件.

class Pool(bootsteps.StartStopStep):
    def create(self, w):
        semaphore = None
        max_restarts = None
        if w.app.conf.worker_pool in GREEN_POOLS:  # pragma: no cover
            warnings.warn(UserWarning(W_POOL_SETTING))
        threaded = not w.use_eventloop or IS_WINDOWS
        procs = w.min_concurrency
        # 果然在这儿
        w.process_task = w._process_task
        """
        省略部分无关代码
        """
        return pool

    def info(self, w):
        return {'pool': w.pool.info if w.pool else 'N/A'}

    def register_with_event_loop(self, w, hub):
        w.pool.register_with_event_loop(hub)

真相大白，自此，我们已经知道了我们的任务是如何被发现，以及如何被消费的了。但是任务究竟是怎么被具体消费的，我们仍然无从知晓。但是我们已经非常幸运的找到了打开这快空白的入口，那就是result = apply_async()。