dramatiq入門

プログラミングwip
最終更新:

dramatiqとは

dramatiqはバックグラウンドタスク処理を行うためのPythonライブラリです．

バックグラウンドタスク処理

アプリケーションでは，計算が多いなどの理由で時間のかかる処理を行うことがあります．このような処理が完了するのを待っていると，使い勝手が悪くなるだけでなく，通信のタイムアウトやサーバーの負荷が高くなるなどの問題が起こります．

これを解決するために，非同期タスクキューを使用します．処理のリクエストはすぐに返し，実際の処理はバックグラウンドで行うことで，ユーザーの待ち時間を減らすことができます． ここで，実際の処理は時間がかかるため，処理のリクエストが渋滞することがあります．また，処理の状態（待機中，実行中，完了など）を管理する必要があります．このような処理の管理を行うのがタスクキューです．

タスクキュー

処理のイメージ

プロデューサーは処理のリクエスト（メッセージ）を作成します．メッセージはキューと呼ばれる待ち行列に追加されます．キューはブローカーと呼ばれる中継役のプログラムによって管理（保存）されます． そして，ワーカーと呼ばれるプログラムがキューからリクエストを取り出してリクエストされた処理を行います．処理はメッセージで指定されたアクターで実行されます．実行結果はバックエンドと呼ばれるプログラムに保存されます．

登場人物をまとめると以下のようになります．

登場人物役割実際のプログラム
プロデューサー処理のリクエスト（メッセージ）を作成するFastAPIなどのWebフレームワーク
ブローカーメッセージを管理するRedisやRabbitMQなどのメッセージブローカー
ワーカー処理を行うdramatiqのワーカープログラム
バックエンド実行結果を保存するRedisやPostgreSQLなどのデータベース

dramatiqはプロデューサーが処理のリクエストを作成するためのAPIと，ワーカーが処理を行うためのAPIを提供し，ブローカーの操作やバックエンドの操作を行います． タスクキューを実装するうえで，ブローカーやバックエンドの操作を意識する必要はありません．

特徴

  • シンプルで使いやすい

    dramatiqはシンプルなAPIを提供しており，簡単にバックグラウンドタスク処理を実装できます．

  • 高い信頼性と性能

    dramatiqは高い信頼性と性能を持っています．タスクの再試行や失敗時の処理など，信頼性を高めるための機能が用意されています．

コード例

アクターの定義

バックエンドで処理を担当するアクターを定義します．

Pythonの関数を作り，その上に@dramatiq.actorという目印を付けます．

python
import dramatiq
from dramatiq.brokers.redis import RedisBroker
import time

# ブローカー（中継役のRedis）の準備
redis_broker = RedisBroker(host="localhost", port=6379)
dramatiq.set_broker(redis_broker)

# アクターの登録（目印をつけるだけ）
@dramatiq.actor
def send_welcome_email(user_email):
    print(f"{user_email} 宛てのメール作成を開始します...")
    
    # 実際にはここで時間のかかる処理（ネットワーク通信など）が行われます
    time.sleep(5) 
    
    print("メールの送信が完了しました！")

プロデューサーでメッセージを追加する

先ほど作った関数に.send()をつけることでブローカーにメッセージが追加されます．

python
from tasks import send_welcome_email

print("ユーザーが登録ボタンを押しました。")

# ワーカーに処理を依頼（キューにメッセージを入れる）
# 普通に呼び出すのではなく .send() を使うのがポイントです
send_welcome_email.send("new_user@example.com")

print("裏方への依頼が一瞬で完了したため、画面はすぐに次のページに進みます。")

dramatiqの起動

dramatiqはこのコマンドで実行します．

bash
dramatiq tasks

処理の結果を受け取る

先ほどのコードでは，処理の結果を受け取ることはできません． バックエンドを用意することで結果を保存できます．

python
import dramatiq
from dramatiq.brokers.redis import RedisBroker
from dramatiq.results import Results
from dramatiq.results.backends.redis import RedisBackend

# 結果保存用のバックエンドを準備
result_backend = RedisBackend()

# ブローカーにバックエンドの仕組み（ミドルウェア）を追加
redis_broker = RedisBroker()
redis_broker.add_middleware(Results(backend=result_backend))
dramatiq.set_broker(redis_broker)

# 結果を保存したいアクターには store_results=True をつける
@dramatiq.actor(store_results=True)
def calculate_tax(price):
    return price * 1.1
python
from tasks import calculate_tax

# 依頼を出す（メッセージオブジェクトが返る）
message = calculate_tax.send(1000)

# ワーカーが処理を終えるまで待機して、実際の実行結果を受け取る
# block=True にすると、処理が終わるまでここで待ちます
result = message.get_result(backend=result_backend, block=True)

print(result) 
# 出力例: 1100.0

このようにバックエンドを設定することで，実行結果を取得することができます．

参考

Dramatiq: background tasks — Dramatiq 2.2.0 documentation
dramatiq.io
RedisとCeleryを用いたバックグラウンド処理を徹底解説: 基本知識から実装まで
zenn.dev