worker啟動
worker的啟動是由celery.bin.worker.worker類中調(diào)用WorkController的start函數(shù)開始的匣吊，其中會調(diào)用Blueprint實例的start函數(shù)燃乍，進而會調(diào)用到Worker的on_start函數(shù)進行一些處理工作。 之后，會遍歷上一步驟Worker實例中創(chuàng)建的steps盒使，調(diào)用它們的start函數(shù)啟動各個組件趾访，至此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函數(shù)，因此這里會調(diào)用到其父類BasePool中定義的函數(shù)阴颖，定義如下

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

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

這里會調(diào)用到on_start函數(shù)活喊，由于各子類覆蓋了該函數(shù)，因此會調(diào)用子類中的on_start函數(shù)量愧，同樣地钾菊，以TaskPool為例，on_start函數(shù)的定義如

作者：薛昊
鏈接：https://zhuanlan.zhihu.com/p/43097583
來源：知乎
著作權(quán)歸作者所有偎肃。商業(yè)轉(zhuǎn)載請聯(lián)系作者獲得授權(quán)煞烫，非商業(yè)轉(zhuǎn)載請注明出處。

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) # 創(chuàng)建Pool

        # Create proxy methods 創(chuàng)建代理
        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函數(shù)主要完成了3個工作
1.根據(jù)選項參數(shù)確定使用BlockingPool還是AsynPool(分別為billiard.pool.Pool和celery.concurrency.asynpool.AsyncPool)
2.創(chuàng)建Pool
3.創(chuàng)建代理方法
關(guān)于AsynPool的具體實現(xiàn)后面再進行分析

Consumer

Consumer與Worker較為類似滞详，都是需要使用BluePrint進行啟動凛俱，詳細流程會在專門分析。這里料饥，我們只簡單瀏覽下Consumer的start函數(shù)

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)