Ubuntu VPSでRabbitMQを使用してCeleryでタスクをキューにする方法

紹介

非同期（ノンブロッキング）処理は、特定のタスクの実行をプログラムのメインフローとは別に分離する方法です。これにより、ユーザー向けのコードを中断することなく実行できるなど、いくつかの利点が提供されます。

メッセージパッシングは、プログラムコンポーネントがコミュニケーションし、情報を交換するために使用できる方法です。これは同期的または非同期的に実装でき、離散プロセスが問題なく通信できるようにします。メッセージパッシングは、この種の使用のために伝統的なデータベースの代替としてしばしば実装されます。なぜなら、メッセージキューはしばしば追加機能を実装し、パフォーマンスを向上させ、完全にメモリ内に存在することができるからです。

セロリは、非同期メッセージパッシングシステムに基づいて構築されたタスクキューです。プログラミングタスクを投げ込むバケツとして使用することができます。タスクを送信したプログラムは、引き続き実行を続け、応答性を保つことができ、後でセロリをポーリングして計算が完了しているかを確認し、データを取得することができます。

セロリはPythonで書かれていますが、そのプロトコルはどの言語でも実装できます。他の言語ともウェブフックを介して機能することができます。

プログラムの環境にジョブキューを実装することで、タスクを簡単にオフロードし、ユーザーとのインタラクションを続けることができます。これは、アプリケーションの応答性を高め、長時間の計算を行っている間にロックされないようにするための簡単な方法です。

このガイドでは、Ubuntu 12.04 VPS上にメッセージングシステムとしてRabbitMQを使用してセレリジョブキューをインストールし実装します。

コンポーネントをインストールする

Celeryをインストールする

セロリはPythonで書かれているため、通常のPythonパッケージを扱うのと同じ方法で簡単にインストールできます。

私たちは、メッセージングシステムをインストールするために仮想環境を作成することにより、Pythonパッケージの処理に関する推奨手順に従います。これにより、環境を安定させ、大規模システムに影響を与えないようにします。

UbuntuのデフォルトリポジトリからPython仮想環境パッケージをインストールします:

sudo apt-get update
sudo apt-get install python-virtualenv

私たちはシステムを実装するメッセージングディレクトリを作成します。

mkdir ~/messaging
cd ~/messaging

以下のコマンドを使用して、セロリをインストールできる仮想環境を今作成できます。

virtualenv --no-site-packages venv

仮想環境が設定されたので、次のように入力してアクティブにできます:

source venv/bin/activate

あなたのプロンプトは、上で作成した仮想環境で操作していることを反映するように変わります。これにより、私たちのPythonパッケージがグローバルではなくローカルにインストールされるようになります。

もし今後、環境を無効にする必要がある場合（今は必要ありません）、次のように入力できます:

deactivate

環境を活性化したので、pipを使ってceleryをインストールできます:

pip install celery

RabbitMQをインストールする

Celeryは外部ソースからのリクエストを処理するためにメッセージエージェントを必要とします。このエージェントは「ブローカー」と呼ばれます。

選択できるブローカーのオプションは多数あります。関係データベース、NoSQLデータベース、キーと値のストア、実際のメッセージングシステムなどが含まれます。

私たちは、堅牢で安定したパフォーマンスを提供し、celeryとうまく連携するRabbitMQメッセージングシステムを使用するようにceleryを設定します。これは、私たちの目的にうまく適合する機能を含んでいるため、素晴らしいソリューションです。

Ubuntuのリポジトリを通じてRabbitMQをインストールできます:

sudo apt-get install rabbitmq-server

RabbitMQサービスは、インストール時に私たちのサーバーで自動的に開始されます。

Celeryインスタンスを作成する

Celeryのタスクキュー機能を使用するためには、インストール後に最初のステップとしてCeleryのインスタンスを作成する必要があります。これは、パッケージをインポートし、「app」を作成し、その後Celeryがバックグラウンドで実行できるタスクを設定するというシンプルなプロセスです。

メッセージングディレクトリ内に、tasks.pyという名前のPythonスクリプトを作成し、そこでワーカーが実行できるタスクを定義しましょう。

sudo nano ~/messaging/tasks.py

最初に行うべきことは、celeryパッケージからCelery関数をインポートすることです:

from celery import Celery

その後、デフォルトのRabbitMQサービスに接続するセレリーアプリケーションのインスタンスを作成できます。

from celery import Celery

app = Celery('tasks', backend='amqp', broker='amqp://')

Celery関数への最初の引数は、タスクを識別するためにその前に付け加えられる名前です。

バックエンドパラメータはオプションのパラメータであり、バックグラウンドタスクの状態をクエリしたり、その結果を取得したい場合には必要です。

もしあなたのタスクが単に作業を行い、その後終了する関数であり、プログラムで使用するための有用な値を返さない場合、このパラメータは省略できます。ただし、タスクの一部だけがこの機能を必要とする場合は、ここで有効にし、さらに必要に応じてケースバイケースで無効にすることができます。

brokerパラメータは、私たちのブローカーに接続するために必要なURLを指定します。私たちの場合、これはサーバー上で実行されているRabbitMQサービスです。RabbitMQは「amqp」というプロトコルを使用して動作します。RabbitMQがデフォルトの設定で動作している場合、celeryはamqp://スキーム以外の情報がなくても接続できます。

Celeryタスクを構築する

このファイルの中で、私たちは今、タスクを追加する必要があります。

各セレリータスクは、デコレーター@app.taskで開始する必要があります。これにより、セレリーはキューイング機能を追加できる関数を識別します。各デコレーターの後に、私たちのワーカーが実行できる関数を作成するだけです。

私たちの最初のタスクは、コンソールに文字列を出力するシンプルな関数です。

from celery import Celery

app = Celery('tasks', backend='amqp', broker='amqp://')

@app.task
def print_hello():
    print 'hello there'

この関数は有用な情報を返さず（代わりにコンソールに出力します）、Celeryに対してこのタスクの状態情報を保存するためにバックエンドを使用しないように指示できます。これは内部の処理が単純になり、リソースも少なくて済みます。

 セロリから Celery をインポートする
app=Celery('tasks', backend='amqp', broker='amqp://')
関数 print_hello() を定義し、'こんにちは' と出力します。
次に、素数を生成する別の関数を追加します（RosettaCodeからの引用）。これは長時間かかるプロセスになる可能性があるため、結果を待っているときに非同期のワーカー処理をどのように扱うかの良い例です。
from celery import Celery

app = Celery('tasks', backend='amqp', broker='amqp://')

@app.task(ignore_result=True)
def print_hello():
    print 'hello there'

@app.task
def gen_prime(x):
    multiples = []
    results = []
    for i in xrange(2, x+1):
        if i not in multiples:
            results.append(i)
            for j in xrange(i*i, x+1, i):
                multiples.append(j)
    return results
この関数の返り値を気にしているため、また、処理が完了したときに結果を使用するため、私たちはこの2番目のタスクにignore_resultパラメータを追加しません。
ファイルを保存して閉じてください。
セレリーワーカープロセスを開始する
私たちは今、アプリケーションからの接続を受け入れることができるワーカープロセスを開始できます。これにより、私たちが作成したファイルを使用して、実行できるタスクについて学習します。
ワーカーインスタンスを開始するのは、celeryコマンドでアプリケーション名を呼び出すだけで簡単です。ワーカープロセスをバックグラウンドに置くために、文字列の最後に「&」文字を付け加えます:
celery worker -A tasks &
これによりアプリケーションが起動し、その後ターミナルから切り離されるので、他の作業に使用し続けることができます。
複数のワーカーを起動したい場合は、-n 引数を使ってそれぞれに名前を付けることで実行できます:
celery worker -A tasks -n one.%h &
celery worker -A tasks -n two.%h &
この%hは、ワーカーの名前が付けられたときにホスト名に置き換えられます。
作業者を停止するには、killコマンドを使用できます。プロセスIDを照会して、この情報に基づいて作業者を排除することができます。
ps auxww | grep 'celery worker' | awk '{print $2}' | xargs kill
これにより、作業者は退出する前に現在のタスクを完了することができます。
すべてのワーカーがタスクを完了するのを待たずにすべてのワーカーをシャットダウンする場合は、以下を実行できます。
ps auxww | grep 'celery worker' | awk '{print $2}' | xargs kill -9
キューを使用して作業を処理する
私たちは、プログラムのためにバックグラウンドで作業を完了するために、生成したワーカー プロセスを使用できます。
この仕組みを示すために全体のプログラムを作成する代わりに、Pythonインタプリタで様々なオプションを探ります:
python
プロンプトで、私たちは関数を環境にインポートできます:
from tasks import print_hello
from tasks import gen_prime
これらの関数をテストすると、特別な機能がないように見えます。最初の関数は期待通りに行を印刷します：
print_hello()
hello there
第二の関数は素数のリストを返します:
primes = gen_prime(1000)
print primes
もし第二の関数にチェックするためのより広い数値の範囲を与えると、計算中に実行が停止します:
primes = gen_prime(50000)
「CTRL-C」を入力して実行を停止します。このプロセスは明らかにバックグラウンドで計算していません。
バックグラウンドワーカーにアクセスするには、.delay メソッドを使用する必要があります。Celery は私たちの関数を追加の機能でラップします。このメソッドは、関数をワーカーに渡して実行するために使用されます。すぐに戻るべきです。
primes = gen_prime.delay(50000)
このタスクは、以前に開始した作業者によって実行されています。アプリケーションのためにbackendパラメーターを設定したため、計算のステータスを確認し、結果にアクセスすることができます。
タスクが完了しているかどうかを確認するために、.ready メソッドを使用できます。
primes.ready()
False
「False」の値は、タスクがまだ実行中であり、結果がまだ利用できないことを意味します。「True」の値を取得した場合、私たちはその答えを使って何かをすることができます。
primes.ready()
True
私たちは.getメソッドを使用することで値を取得できます。
もし私たちがすでにその値が.readyメソッドで計算されていることを確認している場合、そのメソッドを次のように使用できます:
print primes.get()
[2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79, 83, 89, 97, 101, 103, 107, 109, 113, 127, 131, 137, 139, 149, 151, 157, 163, 167, 173, 179, 181, 191, 193, 197, 199, 211, 223, 227, 229, 233, 239, 241, 251, 257, 263, 269, 271, 277, 281, 283, 293, 307, 311, 313, 317, 331, 337, 347, 349, 353, 359, 367, 373, 379, 383, 389, 397, 401, 409, 419, 421, 431, 433, 439, 443, 449, 457, 461, 463, 467, 479, 487, 491, 499, 503, 509, 521, 523,
. . .
しかし、もし.readyメソッドを.getを呼び出す前に使用していない場合、プログラムが結果を待たされることのないように「タイムアウト」オプションを追加することをおそらく望んでいるでしょう。それは私たちの実装の目的に反することになります:
print primes.get(timeout=2)
これがタイムアウトすると例外が発生しますが、それはあなたのプログラムで処理できます。
結論
これは、プログラム内でセロリを使用するために必要な情報としては十分ですが、このライブラリの完全な機能の表面をちょっとなぞるだけです。セロリを使用すると、バックグラウンドタスクを連結したり、タスクをグループ化したり、関数を興味深い方法で組み合わせたりできます。
セロリはPythonで書かれていますが、Webhookを通じて他の言語でも使用することができます。これにより、選択した言語に関係なく、タスクをバックグラウンドに移すことが非常に柔軟になります。
ジャスティン・エリングウッドによる