FastAPIでRabbitMQを使ってみる

# やりたいこと
- FastAPI＋RabbitMQでメッセージを送受信できるようにする
- クライアントからメッセージを送信するためのAPIを作成
- API経由でRabbitMQにメッセージをキューイング
- 受信用サーバでメッセージを受け取る

# デモ
![au1o6-yldc5.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/130670/bba600b9-bd68-0311-5ecc-5a5016f082c6.gif)

# FastAPIとは
2018年12月にリリースされたPython製APIフレームワークです
https://fastapi.tiangolo.com/

特徴としては，Node.jsやGoと同等のパフォーマンス性能でかつ**Flaskライク**な書き方で簡単にAPI構築することができます.
ハッカソンなど時間が限られた開発で有効なフレームワークだと思います.

# RabbitMQとは
何百万ものメッセージを非同期的に仲介することができるオープンソースソフトウェアです．

AMQP（Advanced Message Queuing Protocol）という組織間でビジネスメッセージを送受信するための公開規格（プロトコル）を実装したシステムになります．


# 準備
## 動作に必要なもの
Docker及び，Docker Composeが使えれば大丈夫です！

## ファイル構成

```bash
.
├── consumer
│   ├── Dockerfile
│   ├── consumer.py
│   └── requirements.txt
├── docker-compose.yml
└── producer
    ├── Dockerfile
    ├── producer.py
    └── requirements.txt
```

# コード
## APIサーバ
ヘルスチェック用に`/`と，RabbitMQサーバにメッセージをプッシュするための`/add-job/{message}`の２本を作成します．

RabbitMQへ通信を行うためのパッケージとして**pika**を使用します．

`/add-job/{message}`処理の流れは以下の通りです．

- pikaを使ってRabbitMQサーバと接続
- チャンネルとキューの作成
- 作成したキューに受信したメッセージをプッシュ
- セッションクローズ

```python:./producer/producer.py
from fastapi import FastAPI
import pika

app = FastAPI()


# ヘルスチェック用
@app.get("/")
def read_root():
    return {"Status": "OK"}


# RabbitMQ用
@app.get("/add-job/{message}")
def add_job(message: str):
    # RabbitMQサーバと接続（ホスト名にはコンテナ名を指定しているが，Dockerを使ってない場合はIPアドレスを指定）
    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host="rabbitmq"))
    # チャンネルの確立
    channel = connection.channel()
    # メッセージを格納するためのキュー(task_queue)を作成
    channel.queue_declare(queue="task_queue", durable=True)
    # メッセージをキュー(task_queue)に格納
    channel.basic_publish(
        exchange="",
        routing_key="task_queue",
        body=message,
        properties=pika.BasicProperties(
            delivery_mode=2,  # メッセージの永続化
        ))
    # 接続のクローズ及びメッセージが配信されたことを確認
    connection.close()

    return {"send": message}
```

## 受信サーバ
受信側でもRabbitMQサーバと接続するために**pika**パッケージを利用します．

処理の流れは以下の通りです．

- RabbitMQサーバと接続
- チャンネルの作成＆キューの存在確認
- 受信待機
- 受信時にcallback関数を呼び出し

送信側と主な違いは，メッセージ受取時に**callback関数**が呼び出されることです．

callback関数では，受け取ったメッセージを元に細かに操作することができます．

```python:./consumer/consumer.py
import pika


# メッセージ受信のたびに呼び出される関数
def callback(channel, method, properties, body):
    print(f" [x] Received {body}")
    message = body.decode()

    if message == "hey":
        print("hey there")
    elif message == "hello":
        print("well hello there")
    else:
        print(f"sorry i did not understand {body}")

    print(" [x] Done")
    # 受信したことをキューに知らせる
    channel.basic_ack(delivery_tag=method.delivery_tag)


def main():
    # 初期設定
    print(" [*] Connecting to server ...")
    # RabbitMQサーバと接続（ホスト名にはコンテナ名を指定しているが，Dockerを使ってない場合はIPアドレスを指定）
    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host="rabbitmq"))
    # チャンネルの確立
    channel = connection.channel()
    # メッセージを受信するためのキュー(task_queue)が存在することを確認
    channel.queue_declare(queue="task_queue", durable=True)
    # 前のメッセージの処理が完了してACKが返るまで次のメッセージを送信しないようにするオプション
    channel.basic_qos(prefetch_count=1)
    # キュー(task_queue)にcallback関数をサブスクライブしてメッセージ受信のたびに実行
    channel.basic_consume(queue="task_queue", on_message_callback=callback)

    try:
        print(" [*] Waiting for messages. To exit press Ctrl+C")
        channel.start_consuming()
    except KeyboardInterrupt:
        print(" [x] Done")


if __name__ == '__main__':
    main()
```

## Dockerコンテナ
Docker Composeを使って３つのコンテナ(RabbitMQ，FastAPI，受信サーバ)を立ち上げます．

### APIサーバ
```docker:./producer/Dockerfile
FROM python:3.8-alpine

WORKDIR /app

COPY requirements.txt .
# コンテナ内で必要なパッケージをインストール
RUN apk add --no-cache build-base \
 && pip install --no-cache-dir --trusted-host pypi.python.org -r requirements.txt \
 && apk del build-base

COPY producer.py .
EXPOSE 8000
# FastAPIを8000ポートで待機
CMD ["uvicorn", "producer:app", "--reload", "--host", "0.0.0.0", "--port", "8000"]
```

### 必要パッケージ
```txt:./producer/requirements.txt
uvicorn
fastapi
pika
```

### 受信サーバ
```docker:./consumer/Dockerfile
FROM python:3.8-alpine

WORKDIR /app

COPY requirements.txt .
# コンテナ内で必要なパッケージをインストール
RUN apk add --no-cache build-base \
 && pip install --no-cache-dir --trusted-host pypi.python.org -r requirements.txt \
 && apk del build-base

COPY consumer.py .
```

### 必要パッケージ
```txt:./consumer/requirements.txt
pika
```


### Docker Compose
```yml:./docker-compose.yml
version: '3.0'

services:
  # FastAPI
  producer:
    container_name: producer
    build: ./producer
    restart: always
    tty: true
    ports:
      - 8000:8000
  # Python Client
  consumer:
    container_name: consumer
    build: ./consumer
  # RabbitMQ
  rabbitmq:
    container_name: rabbitmq
    restart: always
    tty: true
    image: rabbitmq:3.8.9-management-alpine
    ports:
      - '5672:5672'
      - '15672:15672'
#    データの永続化が必要な場合は以下をアンコメント
#    volumes:
#      - rabbitmq-data:/var/lib/rabbitmq
#
#volumes:
#  rabbitmq-data:
```

# 実行
## コンテナの構築＆起動
```bash
$ docker-compose up -d --build
```

## コンテナの起動確認
下記コマンドで**RabbitMQ(rabbitmq)**と**FastAPI(producer)**用コンテナがUPステータスであればOKです！

```bash
$ docker-compose ps
  Name                Command               State                                   Ports
------------------------------------------------------------------------------------------------------------------------
consumer   python3                          Exit 0
producer   uvicorn producer:app --rel ...   Up       0.0.0.0:8000->8000/tcp
rabbitmq   docker-entrypoint.sh rabbi ...   Up       15671/tcp, 0.0.0.0:15672->15672/tcp, 15691/tcp, 15692/tcp,
                                                     25672/tcp, 4369/tcp, 5671/tcp, 0.0.0.0:5672->5672/tcp
```

## 受信側の起動
起動に成功すると，` [*] Connecting to server ...`と表示されます．

```bash
$ docker-compose run --rm consumer python consumer.py
Creating sample-rabbitmq_consumer_run ... done
 [*] Connecting to server ...
 [*] Waiting for messages. To exit press Ctrl+C
```

## Swaggerを使ってメッセージを送信
FastAPIでは，APIを作成すると自動でSwaggerドキュメントが生成されます．

今回は，せっかくなのでSwaggerからAPIを叩いてみましょう

### Swaggerへアクセス
http://localhost:8000/docs

### SwaggerからAPIを叩く
ページアクセス後，`/add-job/{message}`を選択->Try it outを選択すると画像のように表示されるので
**message**に**hello**を記入して**Execute**を押すとAPIが叩けます．

![swagger.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/130670/bff66ff7-71f4-80f1-5a6f-ca4ab0188625.png)


### curlを使ってメッセージを送信
もちろん，**curl**を使ってAPIを叩くこともできます．

```bash
$ curl http://localhost:8000/add-job/hello
{"send":"hello"}
```

## メッセージの受け取り
送信と同時に受信側でメッセージが表示されたことがわかります

```bash
 [x] Received b'hello'
well hello there
 [x] Done
```

## RabbitMQ管理画面
RabbitMQコンテナではキューブローカーだけではなく，管理画面も提供されています．
http://localhost:15672/

上記リンクにアクセスすると画像の様な管理画面が表示されます．
![gui.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/130670/6e2d4f8a-5eed-350b-adb9-6015e6b9a0f0.png)

ここでは，時間あたりの受信したメッセージ数やチャンネルの状態など様々な情報を見ることができます．
詳しくは下記記事で説明されているので，気になる方は参照してください．
https://qiita.com/ptiringo/items/c554fa66f0d985394fed

# まとめ
FastAPI＋RabbitMQ環境を構築し，キューイングシステムを使ってメッセージの送受信を実現することができました．
軽量APIフレームワークであるFastAPIは社内システムで使われることが多いので，RabbitMQと合わせるとより出来ることの幅が広がるかもしれません．

# 参考サイト
- Background Processing With RabbitMQ, Python, and Flask<br>
https://medium.com/better-programming/background-processing-with-rabbitmq-python-and-flask-5ca62acf409c
- RabbitMQ tutorial<br>
https://www.rabbitmq.com/tutorials/tutorial-one-python.html