This document describes the current stable version of Celery (3.1). For development docs, go here.

タスク

タスクは Celery アプリケーションの構成要素です。

タスクは関数やメソッドなどの呼び出し可能オブジェクトから生成できるクラスです。タスクは二つの役割を担っています。と言うのも、タスクが呼ばれた時(メッセージ送信時)に起こる内容、それとワーカーがメッセージを受け取った時に起こる内容、この両方を定義しているからです。

すべてのタスククラスは一意な名前を持っており、ワーカーはメッセージに含まれるこの名前を参照して実行すべき関数を見つけます。

タスクメッセージは、ワーカーによって acknowledged されるまで消えることはありません。ワーカーは多くのメッセージを予め保持することができ、電源トラブル等でワーカーが kill されたとしてもメッセージは別のワーカーに再送されるようになっています。

理想的にはタスクの関数は idempotent であるべきです。これは、ある関数が同じ引数で何度呼び出されたとしても意図しない結果を生み出さないことを意味します。ワーカーはタスクが冪等であるかどうか判定できないので、デフォルトの設定では既に開始しているタスクが再実行されないよう、タスクを実行する前にメッセージの確認応答を行います。

もしタスクが冪等である場合は acks_late オプションをセットすると、ワーカーはタスクが完了した 後で メッセージの確認応答を行います。FAQ の Should I use retry or acks_late? も参照してください。

この章ではタスクの定義に関するすべてを学びます。以下が その内容です:

基本

task() デコレーターを使えば、呼び出し可能オブジェクトから簡単にタスクを生成できます。

from .models import User

@app.task
def create_user(username, password):
    User.objects.create(username=username, password=password)

タスクには多くの設定 オプション があり、これらはデコレーターの引数として指定します:

@app.task(serializer='json')
def create_user(username, password):
    User.objects.create(username=username, password=password)

名前

すべてのタスクは一意な名前を持たなければいけません。自分で名前を付けない場合は関数名から新しい名前が生成されます。

例:

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

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

ベストプラクティスはネームスペースとしてモジュール名を使うことです。こうすれば別のモジュールで同じ名前のタスクが既にあったとしても衝突することはありません。

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

タスクの name 属性で名前が分かります:

>>> add.name
'tasks.add'

Which is exactly the name that would have been generated anyway, if the module name is “tasks.py”:

tasks.py:

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

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

自動命名と相対インポート

相対インポートと自動名前生成はうまく共存できません。なので相対インポートを使う場合は明示的に名前をセットするべきです。

例えば、クライアントが “myapp.tasks” モジュールを ”.tasks” とインポートし、ワーカーが “myapp.tasks” とインポートした場合、生成された名前は一致せず、ワーカーから NotRegistered エラーが送出されます。

これは Django の INSTALLED_APPS

INSTALLED_APPS = ['project.myapp']

If you install the app under the name project.myapp then the tasks module will be imported as project.myapp.tasks, so you must make sure you always import the tasks using the same name:

>>> from project.myapp.tasks import mytask   # << GOOD

>>> from myapp.tasks import mytask    # << BAD!!!

The second example will cause the task to be named differently since the worker and the client imports the modules under different names:

>>> from project.myapp.tasks import mytask
>>> mytask.name
'project.myapp.tasks.mytask'

>>> from myapp.tasks import mytask
>>> mytask.name
'myapp.tasks.mytask'

So for this reason you must be consistent in how you import modules, which is also a Python best practice.

Similarly, you should not use old-style relative imports:

from module import foo   # BAD!

from proj.module import foo  # GOOD!

New-style relative imports are fine and can be used:

from .module import foo  # GOOD!

If you want to use Celery with a project already using these patterns extensively and you don’t have the time to refactor the existing code then you can consider specifying the names explicitly instead of relying on the automatic naming:

@task(name='proj.tasks.add')
def add(x, y):
    return x + y

コンテキスト

request は実行しているタスクに関連する情報と状態を保持しています。

リクエストには以下の属性があります:

id:

実行中のタスクのユニーク ID。

group:The unique id a group, if this task is a member.
chord:The unique id of the chord this task belongs to (if the task is part of the header).
args:

引数

kwargs:

キーワード引数

retries:

現在のタスクがリトライされた回数。0 始まりの整数です。

is_eager:

タスクがクライアント内で実行されている場合に True がセットされ、ワーカーによって実行されている場合には False になります。

eta:

(指定されている場合)タスクのオリジナル ETA。UTC 時間です(CELERY_ENABLE_UTC に因る)。

expires:

(指定されている場合)タスクのオリジナル有効期限。UTC 時間です(CELERY_ENABLE_UTC に因る)。

logfile:

ワーカーのログ出力先ファイル。Logging を参照。

loglevel:

現在使用されているログレベル。

hostname:

タスクを実行しているワーカーインタンスのホスト名。

delivery_info:Additional message delivery information. This is a mapping containing the exchange and routing key used to deliver this task. Used by e.g. retry() to resend the task to the same destination queue. Availability of keys in this dict depends on the message broker used.
called_directly:
 

タスクがワーカーによって実行されなかった場合に True。

callbacks:

タスクの実行が成功した場合に呼ばれるサブタスクのリスト。

errback:

タスクの実行が失敗した場合に呼ばれるサブタスクのリスト。

utc:

UTC が有効になっている場合に True (CELERY_ENABLE_UTC)。

バージョン 3.1 で追加.

headers:

メッセージヘッダーのマッピング (おそらく None)。

reply_to:Where to send reply to (queue name).
correlation_id:Usually the same as the task id, often used in amqp to keep track of what a reply is for.

コンテキスト情報にアクセスするタスクの例:

@app.task(bind=True)
def dump_context(self, x, y):
    print('Executing task id {0.id}, args: {0.args!r} kwargs: {0.kwargs!r}'.format(
            self.request))

bind 引数をセットするとその関数は”バウンドメソッド”となり、タスク型インスタンスの属性とメソッドにアクセスできるようになります。

Logging

ワーカーは自動的にログのセットアップをしてくれます。手動で設定することもできます。

A special logger is available named “celery.task”, you can inherit from this logger to automatically get the task name and unique id as part of the logs.

ベストプラクティスは、モジュールの先頭ですべてのタスク共通で使用する logger を生成することです:

from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@app.task
def add(x, y):
    logger.info('Adding {0} + {1}'.format(x, y))
    return x + y

Celery は Python の標準 logger ライブラリを使用します。ドキュメントは logging モジュールのページをご覧ください。

You can also use print(), as anything written to standard out/-err will be redirected to the logging system (you can disable this, see CELERY_REDIRECT_STDOUTS).

注釈

The worker will not update the redirection if you create a logger instance somewhere in your task or task module.

If you want to redirect sys.stdout and sys.stderr to a custom logger you have to enable this manually, for example:

import sys

logger = get_task_logger(__name__)

@app.task(bind=True)
def add(self, x, y):
    old_outs = sys.stdout, sys.stderr
    rlevel = self.app.conf.CELERY_REDIRECT_STDOUTS_LEVEL
    try:
        self.app.log.redirect_stdouts_to_logger(logger, rlevel)
        print('Adding {0} + {1}'.format(x, y))
        return x + y
    finally:
        sys.stdout, sys.stderr = old_outs

リトライ

retry() は回復できるエラーが発生した際などに、タスクを再実行するのに使います。

retry を呼び出すと、同じタスク ID を使って新しいメッセージを元のタスクと同じキューに送信します。

タスクがリトライされたこともタスクのステートに記録されてるので、結果インスタンスを使ってタスクの進行状況を追跡できます(ステート を参照)。

retry を使用する例:

@app.task(bind=True)
def send_twitter_status(self, oauth, tweet):
    try:
        twitter = Twitter(oauth)
        twitter.update_status(tweet)
    except (Twitter.FailWhaleError, Twitter.LoginError) as exc:
        raise self.retry(exc=exc)

注釈

retry() の呼び出しは実際には例外を送出するので、それ以降のコードは実行されません。これは Retry 例外で、エラーとしてではなく、タスクがリトライされたことをワーカーに伝えるための semi-predicate なものとして扱われます。そのため結果バックエンドが有効な場合に正しいステートを保持できます。

This is normal operation and always happens unless the throw argument to retry is set to False.

The bind argument to the task decorator will give access to self (the task type instance).

The exc method is used to pass exception information that is used in logs, and when storing task results. Both the exception and the traceback will be available in the task state (if a result backend is enabled).

If the task has a max_retries value the current exception will be re-raised if the max number of retries has been exceeded, but this will not happen if:

  • An exc argument was not given.

    In this case the MaxRetriesExceeded exception will be raised.

  • There is no current exception

    If there’s no original exception to re-raise the exc argument will be used instead, so:

    self.retry(exc=Twitter.LoginError())
    

    will raise the exc argument given.

Using a custom retry delay

When a task is to be retried, it can wait for a given amount of time before doing so, and the default delay is defined by the default_retry_delay attribute. By default this is set to 3 minutes. Note that the unit for setting the delay is in seconds (int or float).

You can also provide the countdown argument to retry() to override this default.

@app.task(bind=True, default_retry_delay=30 * 60)  # retry in 30 minutes.
def add(self, x, y):
    try:
        …
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)  # override the default and
                                                 # retry in 1 minute

List of Options

The task decorator can take a number of options that change the way the task behaves, for example you can set the rate limit for a task using the rate_limit option.

Any keyword argument passed to the task decorator will actually be set as an attribute of the resulting task class, and this is a list of the built-in attributes.

General

Task.name

The name the task is registered as.

You can set this name manually, or a name will be automatically generated using the module and class name. See 名前.

Task.request

If the task is being executed this will contain information about the current request. Thread local storage is used.

See コンテキスト.

Task.abstract

Abstract classes are not registered, but are used as the base class for new task types.

Task.max_retries

The maximum number of attempted retries before giving up. If the number of retries exceeds this value a MaxRetriesExceeded exception will be raised. NOTE: You have to call retry() manually, as it will not automatically retry on exception..

The default value is 3. A value of None will disable the retry limit and the task will retry forever until it succeeds.

Task.throws

Optional tuple of expected error classes that should not be regarded as an actual error.

Errors in this list will be reported as a failure to the result backend, but the worker will not log the event as an error, and no traceback will be included.

Example:

@task(throws=(KeyError, HttpNotFound)):
def get_foo():
    something()

Error types:

  • Expected errors (in Task.throws)

    Logged with severity INFO, traceback excluded.

  • Unexpected errors

    Logged with severity ERROR, with traceback included.

Task.trail

By default the task will keep track of subtasks called (task.request.children), and this will be stored with the final result in the result backend, available to the client via AsyncResult.children.

This list of task can grow quite big for tasks starting many subtasks, and you can set this attribute to False to disable it.

Task.default_retry_delay

Default time in seconds before a retry of the task should be executed. Can be either int or float. Default is a 3 minute delay.

Task.rate_limit

Set the rate limit for this task type which limits the number of tasks that can be run in a given time frame. Tasks will still complete when a rate limit is in effect, but it may take some time before it’s allowed to start.

If this is None no rate limit is in effect. If it is an integer or float, it is interpreted as “tasks per second”.

The rate limits can be specified in seconds, minutes or hours by appending “/s”, “/m” or “/h” to the value. Tasks will be evenly distributed over the specified time frame.

Example: “100/m” (hundred tasks a minute). This will enforce a minimum delay of 600ms between starting two tasks on the same worker instance.

Default is the CELERY_DEFAULT_RATE_LIMIT setting, which if not specified means rate limiting for tasks is disabled by default.

Note that this is a per worker instance rate limit, and not a global rate limit. To enforce a global rate limit (e.g. for an API with a maximum number of requests per second), you must restrict to a given queue.

Task.time_limit

The hard time limit, in seconds, for this task. If not set then the workers default will be used.

Task.soft_time_limit

The soft time limit for this task. If not set then the workers default will be used.

Task.ignore_result

Don’t store task state. Note that this means you can’t use AsyncResult to check if the task is ready, or get its return value.

Task.store_errors_even_if_ignored

If True, errors will be stored even if the task is configured to ignore results.

Task.send_error_emails

Send an email whenever a task of this type fails. Defaults to the CELERY_SEND_TASK_ERROR_EMAILS setting. See Error E-Mails for more information.

Task.ErrorMail

If the sending of error emails is enabled for this task, then this is the class defining the logic to send error mails.

Task.serializer

A string identifying the default serialization method to use. Defaults to the CELERY_TASK_SERIALIZER setting. Can be pickle json, yaml, or any custom serialization methods that have been registered with kombu.serialization.registry.

Please see シリアライザー for more information.

Task.compression

A string identifying the default compression scheme to use.

Defaults to the CELERY_MESSAGE_COMPRESSION setting. Can be gzip, or bzip2, or any custom compression schemes that have been registered with the kombu.compression registry.

Please see 圧縮 for more information.

Task.backend

The result store backend to use for this task. Defaults to the CELERY_RESULT_BACKEND setting.

Task.acks_late

If set to True messages for this task will be acknowledged after the task has been executed, not just before, which is the default behavior.

Note that this means the task may be executed twice if the worker crashes in the middle of execution, which may be acceptable for some applications.

The global default can be overridden by the CELERY_ACKS_LATE setting.

Task.track_started

If True the task will report its status as “started” when the task is executed by a worker. The default value is False as the normal behaviour is to not report that level of granularity. Tasks are either pending, finished, or waiting to be retried. Having a “started” status can be useful for when there are long running tasks and there is a need to report which task is currently running.

The host name and process id of the worker executing the task will be available in the state metadata (e.g. result.info[‘pid’])

The global default can be overridden by the CELERY_TRACK_STARTED setting.

参考

The API reference for Task.

ステート

Celery can keep track of the tasks current state. The state also contains the result of a successful task, or the exception and traceback information of a failed task.

There are several result backends to choose from, and they all have different strengths and weaknesses (see 結果バックエンド).

During its lifetime a task will transition through several possible states, and each state may have arbitrary metadata attached to it. When a task moves into a new state the previous state is forgotten about, but some transitions can be deducted, (e.g. a task now in the FAILED state, is implied to have been in the STARTED state at some point).

There are also sets of states, like the set of FAILURE_STATES, and the set of READY_STATES.

The client uses the membership of these sets to decide whether the exception should be re-raised (PROPAGATE_STATES), or whether the state can be cached (it can if the task is ready).

You can also define Custom states.

結果バックエンド

If you want to keep track of tasks or need the return values, then Celery must store or send the states somewhere so that they can be retrieved later. There are several built-in result backends to choose from: SQLAlchemy/Django ORM, Memcached, RabbitMQ/QPid (rpc), MongoDB, and Redis – or you can define your own.

No backend works well for every use case. You should read about the strengths and weaknesses of each backend, and choose the most appropriate for your needs.

RPC Result Backend (RabbitMQ/QPid)

The RPC result backend (rpc://) is special as it does not actually store the states, but rather sends them as messages. This is an important difference as it means that a result can only be retrieved once, and only by the client that initiated the task. Two different processes can not wait for the same result.

Even with that limitation, it is an excellent choice if you need to receive state changes in real-time. Using messaging means the client does not have to poll for new states.

The messages are transient (non-persistent) by default, so the results will disappear if the broker restarts. You can configure the result backend to send persistent messages using the CELERY_RESULT_PERSISTENT setting.

Database Result Backend

Keeping state in the database can be convenient for many, especially for web applications with a database already in place, but it also comes with limitations.

  • Polling the database for new states is expensive, and so you should increase the polling intervals of operations such as result.get().

  • Some databases use a default transaction isolation level that is not suitable for polling tables for changes.

    In MySQL the default transaction isolation level is REPEATABLE-READ, which means the transaction will not see changes by other transactions until the transaction is committed. It is recommended that you change to the READ-COMMITTED isolation level.

Built-in States

PENDING

Task is waiting for execution or unknown. Any task id that is not known is implied to be in the pending state.

STARTED

Task has been started. Not reported by default, to enable please see app.Task.track_started.

metadata:pid and hostname of the worker process executing the task.

SUCCESS

Task has been successfully executed.

metadata:result contains the return value of the task.
propagates:Yes
ready:Yes

FAILURE

Task execution resulted in failure.

metadata:result contains the exception occurred, and traceback contains the backtrace of the stack at the point when the exception was raised.
propagates:Yes

RETRY

Task is being retried.

metadata:result contains the exception that caused the retry, and traceback contains the backtrace of the stack at the point when the exceptions was raised.
propagates:No

REVOKED

Task has been revoked.

propagates:Yes

Custom states

You can easily define your own states, all you need is a unique name. The name of the state is usually an uppercase string. As an example you could have a look at abortable tasks which defines its own custom ABORTED state.

Use update_state() to update a task’s state:

@app.task(bind=True)
def upload_files(self, filenames):
    for i, file in enumerate(filenames):
        if not self.request.called_directly:
            self.update_state(state='PROGRESS',
                meta={'current': i, 'total': len(filenames)})

Here I created the state “PROGRESS”, which tells any application aware of this state that the task is currently in progress, and also where it is in the process by having current and total counts as part of the state metadata. This can then be used to create e.g. progress bars.

Creating pickleable exceptions

A rarely known Python fact is that exceptions must conform to some simple rules to support being serialized by the pickle module.

Tasks that raise exceptions that are not pickleable will not work properly when Pickle is used as the serializer.

To make sure that your exceptions are pickleable the exception MUST provide the original arguments it was instantiated with in its .args attribute. The simplest way to ensure this is to have the exception call Exception.__init__.

Let’s look at some examples that work, and one that doesn’t:

# OK:
class HttpError(Exception):
    pass

# BAD:
class HttpError(Exception):

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

# OK:
class HttpError(Exception):

    def __init__(self, status_code):
        self.status_code = status_code
        Exception.__init__(self, status_code)  # <-- REQUIRED

So the rule is: For any exception that supports custom arguments *args, Exception.__init__(self, *args) must be used.

There is no special support for keyword arguments, so if you want to preserve keyword arguments when the exception is unpickled you have to pass them as regular args:

class HttpError(Exception):

    def __init__(self, status_code, headers=None, body=None):
        self.status_code = status_code
        self.headers = headers
        self.body = body

        super(HttpError, self).__init__(status_code, headers, body)

Semipredicates

The worker wraps the task in a tracing function which records the final state of the task. There are a number of exceptions that can be used to signal this function to change how it treats the return of the task.

Ignore

The task may raise Ignore to force the worker to ignore the task. This means that no state will be recorded for the task, but the message is still acknowledged (removed from queue).

This can be used if you want to implement custom revoke-like functionality, or manually store the result of a task.

Example keeping revoked tasks in a Redis set:

from celery.exceptions import Ignore

@app.task(bind=True)
def some_task(self):
    if redis.ismember('tasks.revoked', self.request.id):
        raise Ignore()

Example that stores results manually:

from celery import states
from celery.exceptions import Ignore

@app.task(bind=True)
def get_tweets(self, user):
    timeline = twitter.get_timeline(user)
    if not self.request.called_directly:
        self.update_state(state=states.SUCCESS, meta=timeline)
    raise Ignore()

Reject

The task may raise Reject to reject the task message using AMQPs basic_reject method. This will not have any effect unless Task.acks_late is enabled.

Rejecting a message has the same effect as acking it, but some brokers may implement additional functionality that can be used. For example RabbitMQ supports the concept of Dead Letter Exchanges where a queue can be configured to use a dead letter exchange that rejected messages are redelivered to.

Reject can also be used to requeue messages, but please be very careful when using this as it can easily result in an infinite message loop.

Example using reject when a task causes an out of memory condition:

import errno
from celery.exceptions import Reject

@app.task(bind=True, acks_late=True)
def render_scene(self, path):
    file = get_file(path)
    try:
        renderer.render_scene(file)

    # if the file is too big to fit in memory
    # we reject it so that it's redelivered to the dead letter exchange
    # and we can manually inspect the situation.
    except MemoryError as exc:
        raise Reject(exc, requeue=False)
    except OSError as exc:
        if exc.errno == errno.ENOMEM:
            raise Reject(exc, requeue=False)

    # For any other error we retry after 10 seconds.
    except Exception as exc:
        raise self.retry(exc, countdown=10)

Example requeuing the message:

from celery.exceptions import Reject

@app.task(bind=True, acks_late=True)
def requeues(self):
    if not self.request.delivery_info['redelivered']:
        raise Reject('no reason', requeue=True)
    print('received two times')

Consult your broker documentation for more details about the basic_reject method.

Retry

The Retry exception is raised by the Task.retry method to tell the worker that the task is being retried.

Custom task classes

All tasks inherit from the app.Task class. The run() method becomes the task body.

As an example, the following code,

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

will do roughly this behind the scenes:

class _AddTask(app.Task):

    def run(self, x, y):
        return x + y
add = app.tasks[_AddTask.name]

Instantiation

A task is not instantiated for every request, but is registered in the task registry as a global instance.

This means that the __init__ constructor will only be called once per process, and that the task class is semantically closer to an Actor.

If you have a task,

from celery import Task

class NaiveAuthenticateServer(Task):

    def __init__(self):
        self.users = {'george': 'password'}

    def run(self, username, password):
        try:
            return self.users[username] == password
        except KeyError:
            return False

And you route every request to the same process, then it will keep state between requests.

This can also be useful to cache resources, e.g. a base Task class that caches a database connection:

from celery import Task

class DatabaseTask(Task):
    abstract = True
    _db = None

    @property
    def db(self):
        if self._db is None:
            self._db = Database.connect()
        return self._db

that can be added to tasks like this:

@app.task(base=DatabaseTask)
def process_rows():
    for row in process_rows.db.table.all():
        …

The db attribute of the process_rows task will then always stay the same in each process.

Abstract classes

Abstract classes are not registered, but are used as the base class for new task types.

from celery import Task

class DebugTask(Task):
    abstract = True

    def after_return(self, *args, **kwargs):
        print('Task returned: {0!r}'.format(self.request))


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

Handlers

after_return(self, status, retval, task_id, args, kwargs, einfo)

Handler called after the task returns.

パラメタ:
  • status – Current task state.
  • retval – Task return value/exception.
  • task_id – Unique id of the task.
  • args – Original arguments for the task that returned.
  • kwargs – Original keyword arguments for the task that returned.
  • einfoExceptionInfo instance, containing the traceback (if any).

The return value of this handler is ignored.

on_failure(self, exc, task_id, args, kwargs, einfo)

This is run by the worker when the task fails.

パラメタ:
  • exc – The exception raised by the task.
  • task_id – Unique id of the failed task.
  • args – Original arguments for the task that failed.
  • kwargs – Original keyword arguments for the task that failed.
  • einfoExceptionInfo instance, containing the traceback.

The return value of this handler is ignored.

on_retry(self, exc, task_id, args, kwargs, einfo)

This is run by the worker when the task is to be retried.

パラメタ:
  • exc – The exception sent to retry().
  • task_id – Unique id of the retried task.
  • args – Original arguments for the retried task.
  • kwargs – Original keyword arguments for the retried task.
  • einfoExceptionInfo instance, containing the traceback.

The return value of this handler is ignored.

on_success(self, retval, task_id, args, kwargs)

Run by the worker if the task executes successfully.

パラメタ:
  • retval – The return value of the task.
  • task_id – Unique id of the executed task.
  • args – Original arguments for the executed task.
  • kwargs – Original keyword arguments for the executed task.

The return value of this handler is ignored.

on_retry

How it works

Here comes the technical details, this part isn’t something you need to know, but you may be interested.

All defined tasks are listed in a registry. The registry contains a list of task names and their task classes. You can investigate this registry yourself:

>>> from proj.celery import app
>>> app.tasks
{'celery.chord_unlock':
    <@task: celery.chord_unlock>,
 'celery.backend_cleanup':
    <@task: celery.backend_cleanup>,
 'celery.chord':
    <@task: celery.chord>}

This is the list of tasks built-in to celery. Note that tasks will only be registered when the module they are defined in is imported.

The default loader imports any modules listed in the CELERY_IMPORTS setting.

The entity responsible for registering your task in the registry is the metaclass: TaskType.

If you want to register your task manually you can mark the task as abstract:

class MyTask(Task):
    abstract = True

This way the task won’t be registered, but any task inheriting from it will be.

When tasks are sent, no actual function code is sent with it, just the name of the task to execute. When the worker then receives the message it can look up the name in its task registry to find the execution code.

This means that your workers should always be updated with the same software as the client. This is a drawback, but the alternative is a technical challenge that has yet to be solved.

Tips とベストプラクティス

必要ないなら結果は無視する

タスクの結果を気にしないのであれば、ignore_result オプションをセットしてください。結果の保存は時間とリソースを消費します。

@app.task(ignore_result=True)
def mytask(…):
    something()

CELERY_IGNORE_RESULT 設定を使って結果の保存をグローバルに無効化することもできます。

必要ない場合はレート制限を無効にする

レート制限を使っているタスクがないのであれば、この機能を無効にすることを推奨します。レート制限のサブシステムはとても複雑なものだからです。

CELERY_DISABLE_RATE_LIMITS 設定によってグローバルにレート制限を無効化できます。

CELERY_DISABLE_RATE_LIMITS = True

You find additional optimization tips in the Optimizing Guide.

同期処理のサブタスクは避ける

あるタスクに別のタスクの結果を待たせるのはとても非効率で、ワーカープールが不足するとデッドロックを引き起こしてしまう可能性すらあります。

代わりに非同期処理になるよう設計しましょう、例えば コールバック を使って。

Bad:

@app.task
def update_page_info(url):
    page = fetch_page.delay(url).get()
    info = parse_page.delay(url, page).get()
    store_page_info.delay(url, info)

@app.task
def fetch_page(url):
    return myhttplib.get(url)

@app.task
def parse_page(url, page):
    return myparser.parse_document(page)

@app.task
def store_page_info(url, info):
    return PageInfo.objects.create(url, info)

Good:

def update_page_info(url):
    # fetch_page -> parse_page -> store_page
    chain = fetch_page.s(url) | parse_page.s() | store_page_info.s(url)
    chain()

@app.task()
def fetch_page(url):
    return myhttplib.get(url)

@app.task()
def parse_page(page):
    return myparser.parse_document(page)

@app.task(ignore_result=True)
def store_page_info(info, url):
    PageInfo.objects.create(url=url, info=info)

ここでは異なる subtask() を結合してタスクチェーンを生成しています。Canvas: 設計ワークフロー で、チェーンをはじめ強力なワークフローコンポーネントについて説明しています。

パフォーマンスと戦略

粒度

タスクの粒度はそれぞれのサブタスクが必要とする処理の総量です。一般的には、少ない大きなタスクより、たくさんの小さなタスクに分割した方が良しとされています。

小さなタスクであれば、並列に多くのタスクを実行でき、またワーカーをブロックして別のタスクを待たせたままにすることを避けられます。

しかしながら、タスクの実行にはメッセージの送信だったり、データがローカルになかったりと、オーバーヘッドがあります。そのためタスクを過度に細かく分割してしまうと結果的にこのオーバーヘッド分のコストが上回ってしまうことがあります。

参考

The book Art of Concurrency has a section dedicated to the topic of task granularity [AOC1].

[AOC1]Breshears, Clay. Section 2.2.1, “The Art of Concurrency”. O’Reilly Media, Inc. May 15, 2009. ISBN-13 978-0-596-52153-0.

Data locality

タスクを処理するワーカーは可能な限りデータに近いところにあるべきです。ベストなのはメモリにコピーを持つことで、ワーストなのは別大陸からすべて送ってもらうことです。

もしデータが遠くにあるのなら、そこで別のワーカーを動かすようにします。それができない場合にはよく使うデータをキャッシュする、または使用されることが分かっているデータをプリロードしておくようにします。

複数のワーカー間でデータを共有する一番簡単な方法は、memcached のような分散キャッシュシステムを使うことです。

参考

The paper Distributed Computing Economics by Jim Gray is an excellent introduction to the topic of data locality.

State

Celery は分散システムなので、どのプロセスで、どのマシンでタスクが実行されるか知ることはできません。また、タスクが時間通りに実行されるかについても分かりません。

古めかしい非同期の格言では “asserting the world is the responsibility of the task” と説いています。この意味は、「世界はタスクがリクエストされた時とは変わっているかも知れない。だからタスクは世界があるべき姿であることを確認する責務がある。」ということです。例えば、検索エンジンの再インデックスを行うタスクがあり、その検索エンジンは最大で5分毎にのみ再インデックスされるべきであるという仕様を履行する責務を負うのはタスクの呼び出し側ではなく、タスク自身です。

Django のモデルオブジェクトについて考えてみましょう。このオブジェクトはタスクの引数として渡すべきではありません。一般的にはタスクの中でデータベースからオブジェクトを再取得する方が良しとされる手法です。というのも、古いデータを使うことはレースコンディションを引き起こしてしまうからです。

記事の中に含まれる略語を元の語句に変換するタスクがあったとしましょう:

class Article(models.Model):
    title = models.CharField()
    body = models.TextField()

@app.task
def expand_abbreviations(article):
    article.body.replace('MyCorp', 'My Corporation')
    article.save()

著者はまず記事を作成して保存し、次に略語変換タスクをキックするボタンを押します:

>>> article = Article.objects.get(id=102)
>>> expand_abbreviations.delay(article)

この時、キューは一杯で、このタスクが実行されるまでに2分掛かるとします。その間に別の著者が記事を変更したとすると、タスクは引数として元の古い本文を持っているため、タスクが実行された時に記事の本文は古いものに戻ってしまいます。

このレースコンディションを修正するのは簡単で、引数に記事 ID を使うようにし、タスクの中で記事を再取得すればいいのです:

@app.task
def expand_abbreviations(article_id):
    article = Article.objects.get(id=article_id)
    article.body.replace('MyCorp', 'My Corporation')
    article.save()

>>> expand_abbreviations(article_id)

大きいメッセージを送信するコストも大きいため、このようなアプローチはパフォーマンス的にも有利です。

データベース トランザクション

それでは別の例も見てみましょう:

from django.db import transaction

@transaction.commit_on_success
def create_article(request):
    article = Article.objects.create(…)
    expand_abbreviations.delay(article.pk)

これは Django のビュー関数で記事オブジェクトをデータベースに保存し、その主キーをタスクに渡しています。ここでは commit_on_success デコレーターを使っているため、このビュー関数がリターンしたときにトランザクションをコミットし、例外が送出されたときにはロールバックします。

トランザクションがコミットされる前にタスクの実行が始まってしまうと、レースコンディションが発生します。データベースにまだレコードが存在しないのです!

解決方法としては 現在のトランザクションの状態を前提とするタスクを呼び出す前に、必ずトランザクションをコミットする ことです:

@transaction.commit_manually
def create_article(request):
    try:
        article = Article.objects.create(…)
    except:
        transaction.rollback()
        raise
    else:
        transaction.commit()
        expand_abbreviations.delay(article.pk)

注釈

Django 1.6 (and later) now enables autocommit mode by default, and commit_on_success/commit_manually are deprecated.

This means each SQL query is wrapped and executed in individual transactions, making it less likely to experience the problem described above.

However, enabling ATOMIC_REQUESTS on the database connection will bring back the transaction-per-request model and the race condition along with it. In this case, the simple solution is using the @transaction.non_atomic_requests decorator to go back to autocommit for that view only.

Example

Let’s take a real world example; A blog where comments posted needs to be filtered for spam. When the comment is created, the spam filter runs in the background, so the user doesn’t have to wait for it to finish.

I have a Django blog application allowing comments on blog posts. I’ll describe parts of the models/views and tasks for this application.

blog/models.py

The comment model looks like this:

from django.db import models
from django.utils.translation import ugettext_lazy as _


class Comment(models.Model):
    name = models.CharField(_('name'), max_length=64)
    email_address = models.EmailField(_('email address'))
    homepage = models.URLField(_('home page'),
                               blank=True, verify_exists=False)
    comment = models.TextField(_('comment'))
    pub_date = models.DateTimeField(_('Published date'),
                                    editable=False, auto_add_now=True)
    is_spam = models.BooleanField(_('spam?'),
                                  default=False, editable=False)

    class Meta:
        verbose_name = _('comment')
        verbose_name_plural = _('comments')

In the view where the comment is posted, I first write the comment to the database, then I launch the spam filter task in the background.

blog/views.py

from django import forms
from django.http import HttpResponseRedirect
from django.template.context import RequestContext
from django.shortcuts import get_object_or_404, render_to_response

from blog import tasks
from blog.models import Comment


class CommentForm(forms.ModelForm):

    class Meta:
        model = Comment


def add_comment(request, slug, template_name='comments/create.html'):
    post = get_object_or_404(Entry, slug=slug)
    remote_addr = request.META.get('REMOTE_ADDR')

    if request.method == 'post':
        form = CommentForm(request.POST, request.FILES)
        if form.is_valid():
            comment = form.save()
            # Check spam asynchronously.
            tasks.spam_filter.delay(comment_id=comment.id,
                                    remote_addr=remote_addr)
            return HttpResponseRedirect(post.get_absolute_url())
    else:
        form = CommentForm()

    context = RequestContext(request, {'form': form})
    return render_to_response(template_name, context_instance=context)

To filter spam in comments I use Akismet, the service used to filter spam in comments posted to the free weblog platform Wordpress. Akismet is free for personal use, but for commercial use you need to pay. You have to sign up to their service to get an API key.

To make API calls to Akismet I use the akismet.py library written by Michael Foord.

blog/tasks.py

from celery import Celery

from akismet import Akismet

from django.core.exceptions import ImproperlyConfigured
from django.contrib.sites.models import Site

from blog.models import Comment


app = Celery(broker='amqp://')


@app.task
def spam_filter(comment_id, remote_addr=None):
    logger = spam_filter.get_logger()
    logger.info('Running spam filter for comment %s', comment_id)

    comment = Comment.objects.get(pk=comment_id)
    current_domain = Site.objects.get_current().domain
    akismet = Akismet(settings.AKISMET_KEY, 'http://{0}'.format(domain))
    if not akismet.verify_key():
        raise ImproperlyConfigured('Invalid AKISMET_KEY')


    is_spam = akismet.comment_check(user_ip=remote_addr,
                        comment_content=comment.comment,
                        comment_author=comment.name,
                        comment_author_email=comment.email_address)
    if is_spam:
        comment.is_spam = True
        comment.save()

    return is_spam