やりたいこと
- FastAPI＋RabbitMQでメッセージを送受信できるようにする
- クライアントからメッセージを送信するためのAPIを作成
- API経由でRabbitMQにメッセージをキューイング
- 受信用サーバでメッセージを受け取る
デモ
FastAPIとは
2018年12月にリリースされたPython製APIフレームワークです
https://fastapi.tiangolo.com/
特徴としては，Node.jsやGoと同等のパフォーマンス性能でかつFlaskライクな書き方で簡単にAPI構築することができます.
ハッカソンなど時間が限られた開発で有効なフレームワークだと思います.
RabbitMQとは
何百万ものメッセージを非同期的に仲介することができるオープンソースソフトウェアです．
AMQP（Advanced Message Queuing Protocol）という組織間でビジネスメッセージを送受信するための公開規格（プロトコル）を実装したシステムになります．
準備
動作に必要なもの
Docker及び，Docker Composeが使えれば大丈夫です！
ファイル構成
.
├── 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サーバと接続
- チャンネルとキューの作成
- 作成したキューに受信したメッセージをプッシュ
- セッションクローズ
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関数では，受け取ったメッセージを元に細かに操作することができます．
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サーバ
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"]
必要パッケージ
uvicorn
fastapi
pika
受信サーバ
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 .
必要パッケージ
pika
Docker Compose
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:
実行
コンテナの構築＆起動
$ docker-compose up -d --build
コンテナの起動確認
下記コマンドでRabbitMQ(rabbitmq)とFastAPI(producer)用コンテナがUPステータスであればOKです！
$ 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 ...と表示されます．
$ 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へアクセス
SwaggerからAPIを叩く
ページアクセス後，/add-job/{message}を選択->Try it outを選択すると画像のように表示されるので
messageにhelloを記入してExecuteを押すとAPIが叩けます．
curlを使ってメッセージを送信
もちろん，curlを使ってAPIを叩くこともできます．
$ curl http://localhost:8000/add-job/hello
{"send":"hello"}
メッセージの受け取り
送信と同時に受信側でメッセージが表示されたことがわかります
[x] Received b'hello'
well hello there
[x] Done
RabbitMQ管理画面
RabbitMQコンテナではキューブローカーだけではなく，管理画面も提供されています．
http://localhost:15672/
上記リンクにアクセスすると画像の様な管理画面が表示されます．
ここでは，時間あたりの受信したメッセージ数やチャンネルの状態など様々な情報を見ることができます．
詳しくは下記記事で説明されているので，気になる方は参照してください．
https://qiita.com/ptiringo/items/c554fa66f0d985394fed
まとめ
FastAPI＋RabbitMQ環境を構築し，キューイングシステムを使ってメッセージの送受信を実現することができました．
軽量APIフレームワークであるFastAPIは社内システムで使われることが多いので，RabbitMQと合わせるとより出来ることの幅が広がるかもしれません．
参考サイト
- Background Processing With RabbitMQ, Python, and Flask
https://medium.com/better-programming/background-processing-with-rabbitmq-python-and-flask-5ca62acf409c - RabbitMQ tutorial
https://www.rabbitmq.com/tutorials/tutorial-one-python.html