アジェンダ

そもそもなぜQueue?
AWS SQSの各種用語の紹介
- スタンダードキュー vs FIFOキュー
- 可視性タイムアウト
- MessageGroupId
- 重複排除ID
Pythonにて実際に書いてみよう
- 環境変数周り
- キューの名前からurlを取得する
- メッセージの送信
- メッセージの受信

実行環境

Mac Windowsどちらでも動作確認済み

$ python --version
Python 3.11.4

$ pip list
Package                 Version
----------------------- --------
boto3                   1.28.69

そもそもなぜQueue?

AWSのSQS(Simple Queue Service)に限らず、なぜQueueは必要になるのか、使うことによるメリットはなんなのか。
結論からお話しすると、『Queueの良さは疎結合』 にあると思っています。

疎結合によるメリットはなんなのか。下記のようなメリットがあります！

疎結合による単一障害点の防止
送る側と受け取る側の状況を考える必要がない

従来	SQS

上記よく分からない方向けに、ユースケースで話していきます。

疎結合による単一障害点の防止

サービス間通信をDBを用いて対応した場合、DBに障害が生じると、別サービスにも影響が出てしまいます。
サービス間にキューを置くことでどちらかに障害が発生してもどちらかのサービスは通常通りの挙動を取れます。

送る側と受け取る側の状況を考える必要がない

サービス間通信時に送る側の処理が軽く、受け取る側の処理が重い場合(逆も然り)、処理がどんどん溜まってしまいどちらかの負荷が増えてしまいます。
中間にキューをおいた場合、どちらの状況でも各サービスが対応できる範囲で処理をするだけなので負荷がかかることはありません。

上記のような機会損失を減らせるのがキューの良さだと思っています。

サービスの中間にキューを置くことで、
互いの状況に干渉されず、やることが明確になります。

AWS SQSの各種用語の紹介

ここからはこれだけは覚えておきたいというSQSの用語をざっと紹介していきたいと思います。

SQSに限らずQueueは送る側をプロデューサー / 受け取る側をコンシューマーと言ったりします。
実装時はそこまで気にする必要はない(当記事でも記載していない)ですが、合わせて覚えておくと良いでしょう。

SQSは文字列のみ収納可能(jsonなどは対応可能)

スタンダードキュー vs FIFOキュー

AWS SQSにはQueueのタイプが二つあります。
その一つがスタンダードキューであり、もう一つがFIFOキューです。
FIFOとはfirst in first outの略でして、最初に入ったものを最初に出す -> 順序性が保たれる。になります。

それぞれの特徴をざっとまとめます。

	スタンダード	vs	FIFO
1秒あたりのAPIの制限なし	⭕️	-	❌
順序性の担保	❌	-	⭕️
一回だけ処理する	🔺	-	⭕️

可視性タイムアウト

可視性タイムアウトとは 「受信したキューを一定期間他から見えなくさせる期間」 のことを言います。

例えば可視性タイムアウトを30秒と設定したとしましょう。

受信者Aが一度①というキューを取得した場合、受信者Aもしくは他の受信者は①というキューを30秒間取得でなくなります。

スタンダードキュー / FIFOキューどちらでも設定可能
ただし、スタンダードキューではそうならない可能性もある

MessageGroupId

MessageGroupIdとは、特定のメッセージをグループ分けする際に使用するIDになります。
この値をうまく使うことで、FIFOキューの順序性をさらに細かく制御することができます。

FIFOキューのみ設定可能

重複排除ID

重複排除IDとは、設定したIDと一致するQueueを5分間受け付けなくさせる機能です。

FIFOキューのみ設定可能

Pythonで実際に書いてみよう

ここからは実際にコードを書いてみたいと思います。
参考になれば幸いです。

環境変数周り

環境変数の設定方法ですが、aws cliを使って.awsに認証情報を保持する

もしくは

下記の環境変数を使用し.envファイルに記載すると勝手に読み取ってくれます

// .env
AWS_ACCESS_KEY_ID='xx'
AWS_SECRET_ACCESS_KEY='xxxx'
AWS_DEFAULT_REGION='xxx'

そしてルートの階層でenvファイルを読み込みます。

from dotenv import load_dotenv

load_dotenv()

キューの名前からurlを取得する

URLを環境変数に設定するのがダサいと思ってしまった方向けです。
SQSではキューの名前からurlを取得するAPIがあります。

import boto3

# 引数nameはsqsのキューの名前を指定する
def get_sqs_url(name):
    clientSQS = boto3.client("sqs")
    try:
        res = clientSQS.get_queue_url(QueueName=name)
        return res["QueueUrl"]
    except Exception as e:
        print("err", e)
        return

メッセージの送信

メッセージを送信する場合はsend_messageを使います

import boto3

# 引数にキューのurl, body, 必要な場合はmesage_idを渡す
def push_sqs(url, body, messageId):
    clientSQS = boto3.client("sqs")
    try:
        res = clientSQS.send_message(QueueUrl=url, MessageBody=body, MessageGroupId=mesage_id)
        return res
    except Exception as e:
        print("err", e)
        return

メッセージの受信

メッセージを受信する場合は、receive_messageを使います

import boto3

# 引数にキューのurlを指定します
def get_sqs_data(url):
    clientSQS = boto3.client("sqs")
    try:
        res = clientSQS.receive_message(QueueUrl=url)
        return res
    except Exception as e:
        print("err", e)
        return

メッセージを削除

キューは明示的に削除しない限り残り続けます。(※保存期間が過ぎるまでは残る)
そのため、不要になったキューは削除する必要があります。
削除はdelete_messageを使います。

import boto3

# 引数にキューのurlとキューを取得した際に返却されるreceipt_handleを渡す
def delete_message(url, receipt_handle):
    clientSQS = boto3.client("sqs")
    try:
        res = clientSQS.delete_message(
            QueueUrl=url,
            ReceiptHandle=receipt_handle,
        )
    except Exception as e:
        print("err", e)

とっても簡単でよきですね。

参考