RabbitMQ + PostgreSQL + Celeryで非同期タスク処理をしてみる

# 全体像

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/78249/ce0915e7-5062-caa7-d4a7-4517713926c0.png)

1. クライアントアプリケーションはRabbitMQにタスクを登録する
1. CeleryのワーカーがRabbitMQからタスクを取得し、実行
1. 実行結果をPostgreSQLに保存
1. 実行結果をRabbitMQに返却
1. クライアントアプリケーションはPostgreSQLから実行結果を取得
1. Flowerはワーカーやタスクの実行状況を監視

# 使用技術

### RabbitMQ

- メッセージキューを実現するためのOSSで、メッセージブローカと呼ばれる
- 詳しくは[Docker Composeで立てたRabbitMQにpikaで接続してみる](https://qiita.com/h_tyokinuhata/items/1097a3a0ad018366ecec)を参照

### Celery

- Pythonで非同期タスク処理を行うライブラリ
- RabbitMQ以外にもRedis等、様々なメッセージブローカに対応している
- GitHub: https://github.com/celery/celery

### Flower

- Celery専用のモニタリングツール
- GitHub: https://github.com/mher/flower

### PostgreSQL

- いわずもがなRDBMS
- 今回は実行結果の永続化に使用する
- 実行結果を永続化しない場合、クライアントアプリケーションから実行結果を取得できなくなる

# 環境構築

### Docker Composeの設定

- RabbitMQとPostgreSQLのDockerコンテナを起動する

```yaml:compose.yml
services:

  rabbitmq:
    image: rabbitmq:3.11.7-management
    container_name: rabbitmq
    ports:
      - 5672:5672
      - 15672:15672
    environment:
      - RABBITMQ_DEFAULT_USER=root
      - RABBITMQ_DEFAULT_PASS=password
    volumes:
      - ./docker/rabbitmq/data:/var/lib/rabbitmq

  postgresql:
    image: postgres:latest
    container_name: postgresql
    ports:
      - 5432:5432
    environment:
      - POSTGRES_DB=celery
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - PGDATA=/var/lib/postgresql/data/pgdata
    volumes:
      - ./docker/postgresql/data:/var/lib/postgresql/data
```

```bash
# Dockerコンテナの起動
$ docker compose up -d

# Dockerコンテナが起動していることを確認
$ docker compose ps
NAME                COMMAND                  SERVICE             STATUS              PORTS
postgresql          "docker-entrypoint.s…"   postgresql          running             0.0.0.0:5432->5432/tcp
rabbitmq            "docker-entrypoint.s…"   rabbitmq            running             0.0.0.0:5672->5672/tcp, 0.0.0.0:15672->15672/tcp

# RabbitMQに接続できることを確認
$ open http://localhost:15672

# PostgreSQLに接続できることを確認
$ psql -h localhost -p 5432 -U postgres -d celery
Password for user postgres: 
psql (14.5 (Homebrew), server 15.2 (Debian 15.2-1.pgdg110+1))
WARNING: psql major version 14, server major version 15.
         Some psql features might not work.
Type "help" for help.

celery=# 
```

### ライブラリのインストール

```bash
$ pipenv install celery
$ pipenv install flower

# CeleryからPostgreSQLを操作するために必要
$ pipenv install SQLAlchemy
$ pipenv install psycopg2
```

### Flowerの起動

```bash
$ celery --broker=amqp://root:password@localhost:5672// flower
$ open localhost:5555
```

![FireShot Capture 157 - Flower - localhost.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/78249/1324ad6f-9988-4590-ec5c-c0f5abd92edb.png)

### タスクの実装

```python:tasks.py
from celery import Celery

# brokerにはRabbitMQ、backendにはPostgreSQLの接続情報を記述する
celery = Celery(
    "tasks",
    broker="amqp://root:password@localhost:5672//",
    backend="db+postgresql://postgres:postgres@localhost:5432/celery",
)

# 名前を受け取って挨拶を返すタスク
# Celery.taskデコレータを付与することで、Celeryのタスクとして登録される
@celery.task
def greet(name: str):
    return "hello, {name}!".format(name=name)
```

### ワーカーの起動

```bash
$ celery -A tasks worker --loglevel=INFO

# ワーカーの並列数を指定する場合
$ celery -A tasks worker --concurrency 4 --loglevel=INFO
```

- ワーカーの並列数を指定した場合、Flower上の「Max concurrency」が指定した値になっている
    - ちなみにデフォルトの並列数は`8`

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/78249/4c0f0bbe-ab94-2634-c0b6-166943a8554e.png)

### タスクの実行

```python:client.py
from tasks import greet

# タスクの実行
res = greet.delay("takashi")

# タスクの実行結果の取得
msg = res.get(timeout=1)
print(msg)
```

```bash
$ python client.py
hello, takashi!
```

### FlowerとPostgreSQLの確認

- Flower上の「Processed」と「Succeeded」が増加している

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/78249/8c8bb202-2eee-159f-de86-c8c87a6361f0.png)

- PostgreSQL上に実行結果を保存するテーブルが作成されている

```bash
$ psql -h localhost -p 5432 -U postgres -d celery
Password for user postgres: 
psql (14.5 (Homebrew), server 15.2 (Debian 15.2-1.pgdg110+1))
WARNING: psql major version 14, server major version 15.
         Some psql features might not work.
Type "help" for help.

celery=# \dt
               List of relations
 Schema |        Name        | Type  |  Owner   
--------+--------------------+-------+----------
 public | celery_taskmeta    | table | postgres
 public | celery_tasksetmeta | table | postgres
(2 rows)

```