Go for it!

Debian Lennyでcelery/rabbitmqを動かすメモ

セロリ、ニンジン、ウサギが好きです（嘘です）。

loc2go.comはTCX→KMLのデータコンバートをオンライン＝同期で行っているのですが、サーバは重いしクライアントを待たせすぎ（使ってる方、重くてスイマセン）なのでジョブキュー設計を導入したいと考えてました。

で、いろいろ調べてみるとceleryとRabbitMQが面白そうなのでちょっと試してみた。

ざっくりとしたシステム概要は、celeryがpythonで書かれたキューマネージャ、バックエンドにはerlangで書かれたRabbitMQ+適当なRDBMSが置かれる。タスク依頼はAMQPを使ってceleryに依頼し、RabbitMQがキューをワーカへ投げる（という理解で合ってるんだろうか）。

vmwareにDebian Lennyの仮想マシンを作ってちょこちょこと。

aptで以下を入れる。

• python2.5
• python-setuptools
• python-django
• erlang-nox

コマンドだとこんなかんじ。

# aptitude install python2.5 python-setuptools python-django erlang-nox

RabbitMQはdebがあったので/etc/apt/source.listに以下を追加。

deb http://www.rabbitmq.com/debian/ testing main

aptでガツンと入れる。

# aptitude install rabbitmq-server

チュートリアル通りにRabbitMQの認証設定をする。

$ rabbitmqctl add_user myuser mypassword
$ rabbitmqctl add_vhost myvhost
$ rabbitmqctl set_permissions -p myvhost myuser "" ".*" ".*"

celeryとmultiprocessing入れる（pip入ってなかったらここから入れてください）。

# pip install celery
# pip install multiprocessing

djangoのプロジェクトを作る。

$ mkdir ~/project/
$ cd ~/project/
$ django-admin.py startproject celerytest
$ cd celerytest

INSTALLED_APPSにceleryを追加。データベースは適当に設定します（私はsqlite3を使いました）。

INSTALLED_APPS = (
  'django.contrib.auth',
  'django.contrib.contenttypes',
  'django.contrib.sessions',
  'django.contrib.sites',
  'celery', #  これを追加。
  'mytask', # これも追加（後述）
}

AMQPサーバの設定も加えます。

AMQP_SERVER='127.0.0.1'
AMQP_PORT=5672
AMQP_USER='myuser'
AMQP_PASSWORD='mypassword'
AMQP_VHOST='myvhost'

で、データベースを作るとジョブ管理用のスキーマができあがります。

$ python manage.py syncdb

非同期タスクはcelery.task.Taskを継承して作る。タスクはdjangoアプリケーションで作成する必要があるようなので、mytaskというアプリケーションを作る。

$ python manage.py startapp mytask

mytask/tasks.pyに以下を書く。内容は「5秒待ってhogehogeと返す」だけ。

from celery.task import Task
from celery.registry import tasks
import time

class MyTask(Task):
  name='MyTask'

  def run(self, **kwargs):
    time.sleep(5)
    return 'hogehoge'

tasks.register(MyTask)

celerydはmanage.pyから起動する。タスクはceleryd起動時に読み込まれる。

$ python manage.py celeryd

アプリケーションがあればipythonからタスクを呼び出せる。tasksをインポート。

from mytask import tasks

delayメソッドを呼び出すとタスクが走る（他にもあるけどよく知らん）。

res=tasks.MyTask.delay()

resにAsyncResultオブジェクトが入る。

print res
<AsyncResult: c50cc259-ab02-46be-8ec6-e5c63ce4e1e2>

タスクが完了しているかどうかはsuccessful()メソッドを使う。完了していなければFalseが返る。

print res.successful()
False

5秒後に再度実行するとTrueが返る、はず。

print res.successful()
True

値はget()メソッドを使う。

print res.get()
'hogehoge'

ちょっと駆け足でしたがPython Hack-a-thon #2で熱いネタだったのでとりあえず公開。突っ込み歓迎。
–
もしmultiprocessing周りでエラーが出たら、以下のURLのパッチを適用する。
http://robertpogorzelski.com/blog/2009/09/10/rabbitmq-celery-and-django/

–
is_done()なくなるよ！と指摘をいただいたので修正した。(2009/11/15 1:09)