Celery(三)實例Application

Celery必須實例化后才可以使用，實例稱之為application或者簡稱app。實例是線程安全的，多個Celery實例(不同的配置、部件和任務)都可以在一個進程空間中運行。

創建一個最簡單的app：

>>> from celery import Celery
>>> app = Celery()
>>> app
<Celery __main__ at 0x7f6be52d0cd0>

上述的app是一個運行在__main__模塊中的Celery實例。

Main Name

Celery發送任務信息時，是不發送任何源代碼的，只是發送要執行的任務名稱，而每個worker都維持着一個任務名稱到任務具體函數的映射，稱之為任務注冊。

所以每個任務task都要有一個獨特的不重復名稱，可以看下任務默認名稱是如何生成的：

>>> @app.task
... def add(x, y):
...     return x + y
... 
>>> add.name
u'__main__.add'

可見任務的名稱是實例運行模塊的名稱加上任務函數的名稱。

現在在py文件中創建一個app實例，tasks.py：

from celery import Celery
app = Celery()


@app.task
def add(x, y): return x + y


if __name__ == '__main__':
    print add.name
    app.worker_main()

在shell中直接創建Celery實例、模塊直接運行或者在命令行中運行模塊，都是在main模塊中運行的：

$ python tasks.py
__main__.add

而使用import導入模塊的時候，main name為定義Celery實例模塊的名稱：

>>> from tasks import add
>>> add.name
u'tasks.add'

在main模塊中運行是可以手動指定實例的Main name的：

>>> from celery import Celery
>>> app = Celery('tasks')
>>> app.main
'tasks'

任務的名稱也可以指定：

>>> @app.task(name='sum-of-two-numbers')
>>> def add(x, y):
...     return x + y

>>> add.name
'sum-of-two-numbers'

Configuration

要為app實例添加配置有幾種方式：

創建app實例時初始化：

app = Celery('tasks', backend='redis://localhost:6379/0',
    ┆   ┆   ┆broker='redis://localhost:6379/0')

使用app.conf屬性設置：

app.conf.result_backend = 'redis://localhost:6379/0'
app.conf.broker_url = 'redis://localhost:6379/0'

update多個配置：

>>> app.conf.update(
...     enable_utc=True,
...     timezone='Asia/Shanghai',
...)

使用配置文件，在當前目錄下或者python可以搜索到的目錄下建立一個配置文件，保證可以import，celeryconfig.py ：

result_backend = 'redis://localhost:6379/0'
broker_url = 'redis://localhost:6379/0'

然后：

app.config_from_object('celeryconfig')

可以測試一下配置文件是否有格式錯誤：

$ python -m celeryconfig

也可以建立一個配置類：

class Config:
    enable_utc = True
    timezone = 'Europe/London'

app.config_from_object(Config)

從環境變量中獲取：

import os
from celery import Celery

#: Set default configuration module name
os.environ.setdefault('CELERY_CONFIG_MODULE', 'celeryconfig')

app = Celery()
app.config_from_envvar('CELERY_CONFIG_MODULE')

查看配置：

>>> app.conf.humanize(with_defaults=False, censored=True)
>>> app.conf.table(with_defaults=False, censored=True)

第一條以字符串的形式返回，第二條以字典的形式返回。

with_defaults設置為True時可以查看默認的配置，censored設置為True可以過濾掉敏感信息，包括API, TOKEN, KEY, SECRET, PASS, SIGNATURE, DATABASE。

 Laziness

app實例是延遲的，創建一個實例只會將app設置為current app，只有在真正需要的時候才會完成。

實例只有在調用app.finalize()方法或者訪問app.tasks屬性時才會完成。

Finalizing實例會復制apps之間可以共享的tasks，執行未確定的tasks裝飾器，確定所有的tasks都綁定於current app。

app.task裝飾器並不會真正的創建task任務，直到task被調用或者app finalize完成時才創建：

>>> from celery import Celery
>>> app = Celery()
>>> @app.task
... def add(x, y):
...     return x + y
... 
>>> add.__evaluated__()
False
>>> repr(add)
'<@task: __main__.add of __main__ at 0x7f6571694cd0>'
>>> add.__evaluated__()
True

調用了task的__repr__方法后，task就被真正創建了。

Breaking the chain

當使用current app時，最好將其作為參數進行傳遞，稱之為app chain，比較好的實踐是：

class Scheduler(object):

    def __init__(self, app):
        self.app = app

而不要：

from celery import current_app

class Scheduler(object):

    def run(self):
        app = current_app

通用的的寫法：

from celery.app import app_or_default

class Scheduler(object):
    def __init__(self, app=None):
        self.app = app_or_default(app)

開發時可以設置：

$ CELERY_TRACE_APP=1 celery worker -l info

當app chain斷裂時會raise一個異常。

Abstract Tasks

使用app.task裝飾器創建的任務都繼承自Task類。

可以自定義：

from celery import Task

class DebugTask(Task):

    def __call__(self, *args, **kwargs):
        print('TASK STARTING: {0.name}[{0.request.id}]'.format(self))
        return super(DebugTask, self).__call__(*args, **kwargs)

然后用base參數指定：

@app.task(base=DebugTask)
def add(x, y):
    return x + y

也可以使用app的Task參數修改：

>>> app.Task = DebugTask