celery 源码笔记（三） worker

worker启动
worker的启动是由celery.bin.worker.worker类中调用WorkController的start函数开始的，其中会调用Blueprint实例的start函数，进而会调用到Worker的on_start函数进行一些处理工作。之后，会遍历上一步骤Worker实例中创建的steps，调用它们的start函数启动各个组件，至此worker启动完成
。

image.png

Hub
同样地，首先我们先来看Hub类的start

class Hub(bootsteps.StartStopStep):
    """Worker starts the event loop."""

    def start(self, w):
        pass

可以看到Hub类中并未做任何处理

Pool
这里我们以TaskPool为例（除AsyncPool外其他Pool类似），由于TaskPool中没有声明start函数，因此这里会调用到其父类BasePool中定义的函数，定义如下

class BasePool(object):
    """Task pool."""

    def start(self):
        self._does_debug = logger.isEnabledFor(logging.DEBUG)
        self.on_start()
        self._state = self.RUN

这里会调用到on_start函数，由于各子类覆盖了该函数，因此会调用子类中的on_start函数，同样地，以TaskPool为例，on_start函数的定义如

作者：薛昊
链接：https://zhuanlan.zhihu.com/p/43097583
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

lass TaskPool(BasePool):
    """Multiprocessing Pool implementation."""

    Pool = AsynPool
    BlockingPool = BlockingPool

    uses_semaphore = True
    write_stats = None

    def on_start(self):
        forking_enable(self.forking_enable)
        Pool = (self.BlockingPool if self.options.get('threads', True)
                else self.Pool) # 若使用多线程则使用BlockingPool否则使用AsynPool
        P = self._pool = Pool(processes=self.limit,
                              initializer=process_initializer,
                              on_process_exit=process_destructor,
                              enable_timeouts=True,
                              synack=False,
                              **self.options) # 创建Pool

        # Create proxy methods 创建代理
        self.on_apply = P.apply_async
        self.maintain_pool = P.maintain_pool
        self.terminate_job = P.terminate_job
        self.grow = P.grow
        self.shrink = P.shrink
        self.flush = getattr(P, 'flush', None)  # FIXME add to billiard

可以看到，on_start函数主要完成了3个工作
1.根据选项参数确定使用BlockingPool还是AsynPool(分别为billiard.pool.Pool和celery.concurrency.asynpool.AsyncPool)
2.创建Pool
3.创建代理方法
关于AsynPool的具体实现后面再进行分析

Consumer

Consumer与Worker较为类似，都是需要使用BluePrint进行启动，详细流程会在专门分析。这里，我们只简单浏览下Consumer的start函数

class Consumer(object):
    """Consumer blueprint."""

    def start(self):
        blueprint = self.blueprint
        while blueprint.state not in STOP_CONDITIONS:
            maybe_shutdown()
            if self.restart_count:
                try:
                    self._restart_state.step()
                except RestartFreqExceeded as exc:
                    crit('Frequent restarts detected: %r', exc, exc_info=1)
                    sleep(1)
            self.restart_count += 1
            try:
                blueprint.start(self)
            except self.connection_errors as exc:
                # If we're not retrying connections, no need to catch
                # connection errors
                if not self.app.conf.broker_connection_retry:
                    raise
                if isinstance(exc, OSError) and exc.errno == errno.EMFILE:
                    raise  # Too many open files
                maybe_shutdown()
                if blueprint.state not in STOP_CONDITIONS:
                    if self.connection:
                        self.on_connection_error_after_connected(exc)
                    else:
                        self.on_connection_error_before_connected(exc)
                    self.on_close()
                    blueprint.restart(self)

最后编辑于：2020.08.27 14:14:04