SQS/SNS連携で本番で踏んだ地雷と、3年かけて辿り着いた正解
SNSのFan-outパターンって教科書的には正しいはずなのに、本番運用してみると地雷だらけ。重複メッセージ、配信遅延、サブスクリプションの落とし穴。チームが実際に失敗した例と、それで学んだ設計パターンをまとめました。
SQS/SNS連携パターン、本番で学んだ設計の失敗と正解
先日チームで本番のメッセージング設計を見直すことになって、SQS/SNS連携のパターンについて改めて整理し直したんですよ。正直、これまで「とりあえずFan-outしておけばいい」くらいの理解だったんですが、実際に運用してみると地雷がいっぱい埋まってた。今回は、3年本番で踏んだ失敗と、それで辿り着いた設計パターンについて書きます。
最初の間違い：「Fan-outなら何でもOK」だった
うちのチームが最初にハマったのは、SNSのFan-outパターンを導入した当初の話です。複数のマイクロサービスに同じイベントを配信したいから、SNSトピックに複数のSQSキューをサブスクライブさせる——確かに教科書的には正しいんですよね。
でも実運用で問題が出ました。あるサービスのSQS処理が遅延すると、他のサービスのキューまで圧迫される。それに、SNSから配信されたメッセージが、あるキューでは処理されるのに、別のキューでは「メッセージが来ていない」みたいな不可解なバグが起きたんです。
これ、実はSNSのサブスクリプション設定の問題だったんです。デフォルトでは、サブスクリプション作成時点でのメッセージだけが配信されるので、後から追加したキューには過去のイベントが届かない。また、SNS側でのリトライ戦略とSQS側での失敗処理が完全にズレていて、重複メッセージが大量に溜まるという地獄を経験しました。
# ❌ これが最初の失敗パターン
# SNSトピックに複数SQSをサブスクライブ
# でも、各サービスの処理速度が違うと…
import boto3
import json
from datetime import datetime
sns = boto3.client('sns')
sqs = boto3.client('sqs')
# イベント発行
def publish_event(event_data):
sns.publish(
TopicArn='arn:aws:sns:ap-northeast-1:xxx:user-events',
Message=json.dumps(event_data),
MessageAttributes={
'event_type': {'DataType': 'String', 'StringValue': 'user.created'}
}
)
# 受信側
def consume_from_queue(queue_url):
while True:
response = sqs.receive_message(
QueueUrl=queue_url,
MaxNumberOfMessages=10,
WaitTimeSeconds=20
)
if 'Messages' not in response:
continue
for message in response['Messages']:
# 処理が長いサービスAと短いサービスBがいると…
process_heavy_task() # 30秒かかる
sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=message['ReceiptHandle'])このコードの何が問題かというと、メッセージの処理時間が異なるサービス間で、SNSのFan-outを使うと、遅いサービスがボトルネックになるんです。SNS側ではリトライが発火したり、可視性タイムアウト（VisibilityTimeout）の設定が合致していなかったり、DLQへの送信ルールがバラバラだったり——もう地獄絵図です。
本当に必要だった3つの分岐
3年本番運用で気づいたのは、SQS/SNS連携には「3つの異なるユースケース」があるということです。
1. Fan-out型：複数のサービスが同じイベントを必要とする場合
これが一般的な使い方。ユーザー登録イベントが発火したら、メール送信、プロフィール初期化、分析サービスなど複数のサービスが独立して動く。
ここで重要なのは、各サービスが自分のペースで処理できる独立したキューを持つってことなんですよ。SNS→SQS構成で、各SQSキューの設定を完全に独立させるんです。
# ✅ 改善版：Fan-outの正しい実装
# 各サービスに独立したSQSを用意
import boto3
import json
import logging
from concurrent.futures import ThreadPoolExecutor
import time
sns = boto3.client('sns')
sqs = boto3.client('sqs')
logger = logging.getLogger(__name__)
class EventPublisher:
def __init__(self, topic_arn):
self.topic_arn = topic_arn
def publish(self, event_type, data):
"""イベント発行（SNS）"""
try:
sns.publish(
TopicArn=self.topic_arn,
Message=json.dumps({
'event_type': event_type,
'data': data,
'timestamp': int(time.time())
}),
MessageAttributes={
'event_type': {
'DataType': 'String',
'StringValue': event_type
}
}
)
logger.info(f"Event published: {event_type}")
except Exception as e:
logger.error(f"Failed to publish: {e}")
raise
class QueueConsumer:
def __init__(self, queue_url, dlq_url=None):
self.queue_url = queue_url
self.dlq_url = dlq_url
self.max_retries = 3
def consume(self, handler_func, max_messages=10):
"""キューからメッセージを消費"""
while True:
try:
response = sqs.receive_message(
QueueUrl=self.queue_url,
MaxNumberOfMessages=min(max_messages, 10),
WaitTimeSeconds=20,
AttributeNames=['ApproximateReceiveCount']
)
if 'Messages' not in response:
time.sleep(1)
continue
for message in response['Messages']:
self._process_message(message, handler_func)
except Exception as e:
logger.error(f"Error consuming messages: {e}")
time.sleep(5) # Backoff
def _process_message(self, message, handler_func):
"""メッセージ処理（エラーハンドリング付き）"""
receipt_handle = message['ReceiptHandle']
receive_count = int(message['Attributes']['ApproximateReceiveCount'])
try:
body = json.loads(message['Body'])
logger.info(f"Processing message: {body['event_type']}")
# ハンドラー実行
handler_func(body)
# 成功時は削除
sqs.delete_message(QueueUrl=self.queue_url, ReceiptHandle=receipt_handle)
logger.info(f"Message processed successfully")
except Exception as e:
logger.error(f"Handler error (attempt {receive_count}): {e}")
# リトライ回数超過したらDLQへ
if receive_count >= self.max_retries:
if self.dlq_url:
self._send_to_dlq(message, e)
# 元のキューから削除
sqs.delete_message(QueueUrl=self.queue_url, ReceiptHandle=receipt_handle)
else:
# VisibilityTimeoutを延長して再試行
sqs.change_message_visibility(
QueueUrl=self.queue_url,
ReceiptHandle=receipt_handle,
VisibilityTimeout=60 * (2 ** (receive_count - 1)) # Exponential backoff
)
def _send_to_dlq(self, message, error):
"""DLQへメッセージを転送"""
try:
sqs.send_message(
QueueUrl=self.dlq_url,
MessageBody=json.dumps({
'original_message': json.loads(message['Body']),
'error': str(error),
'timestamp': int(time.time())
})
)
logger.warning(f"Message sent to DLQ")
except Exception as e:
logger.error(f"Failed to send to DLQ: {e}")
# 使用例
if __name__ == "__main__":
publisher = EventPublisher('arn:aws:sns:ap-northeast-1:xxx:user-events')
# イベント発行
publisher.publish('user.created', {'user_id': 123, 'email': 'user@example.com'})
# 各サービスが独立したキューで消費
def email_handler(event):
print(f"Sending email to {event['data']['email']}")
time.sleep(2) # 時間がかかる処理
consumer = QueueConsumer(
'https://sqs.ap-northeast-1.amazonaws.com/xxx/email-queue',
'https://sqs.ap-northeast-1.amazonaws.com/xxx/email-dlq'
)
consumer.consume(email_handler)2. 優先度キュー型：処理の優先度が異なる場合
これは意外と需要があるんですよ。例えば、有料ユーザーのイベントは高優先度で、無料ユーザーは低優先度——こういうときはSNS側で振り分けて、別々のSQSキューに送信するんです。
# CloudFormation でSNS → 複数SQS の優先度分け
AWSTemplateFormatVersion: '2010-09-09'
Resources:
UserEventsTopic:
Type: AWS::SNS::Topic
Properties:
TopicName: user-events
MessageRetentionPeriod: 1209600 # 14日
HighPriorityQueue:
Type: AWS::SQS::Queue
Properties:
QueueName: user-events-high-priority
VisibilityTimeout: 300
MessageRetentionPeriod: 1209600
RedrivePolicy:
deadLetterTargetArn: !GetAtt HighPriorityDLQ.Arn
maxReceiveCount: 3
HighPriorityDLQ:
Type: AWS::SQS::Queue
Properties:
QueueName: user-events-high-priority-dlq
MessageRetentionPeriod: 1209600
LowPriorityQueue:
Type: AWS::SQS::Queue
Properties:
QueueName: user-events-low-priority
VisibilityTimeout: 600 # より長く設定
MessageRetentionPeriod: 1209600
RedrivePolicy:
deadLetterTargetArn: !GetAtt LowPriorityDLQ.Arn
maxReceiveCount: 5 # より多く許可
LowPriorityDLQ:
Type: AWS::SQS::Queue
Properties:
QueueName: user-events-low-priority-dlq
MessageRetentionPeriod: 1209600
# SNS → High Priority Queue のサブスクリプション
HighPrioritySubscription:
Type: AWS::SNS::Subscription
Properties:
Protocol: sqs
TopicArn: !Ref UserEventsTopic
Endpoint: !GetAtt HighPriorityQueue.Arn
FilterPolicy:
user_tier:
- premium
- enterprise
# SNS → Low Priority Queue のサブスクリプション
LowPrioritySubscription:
Type: AWS::SNS::Subscription
Properties:
Protocol: sqs
TopicArn: !Ref UserEventsTopic
Endpoint: !GetAtt LowPriorityQueue.Arn
FilterPolicy:
user_tier:
- free
- trial
# SQS がSNSからメッセージを受け取れるポリシー
HighPriorityQueuePolicy:
Type: AWS::SQS::QueuePolicy
Properties:
Queues:
- !Ref HighPriorityQueue
PolicyText:
Statement:
- Effect: Allow
Principal:
Service: sns.amazonaws.com
Action: sqs:SendMessage
Resource: !GetAtt HighPriorityQueue.Arn
Condition:
ArnEquals:
aws:SourceArn: !Ref UserEventsTopic
LowPriorityQueuePolicy:
Type: AWS::SQS::QueuePolicy
Properties:
Queues:
- !Ref LowPriorityQueue
PolicyText:
Statement:
- Effect: Allow
Principal:
Service: sns.amazonaws.com
Action: sqs:SendMessage
Resource: !GetAtt LowPriorityQueue.Arn
Condition:
ArnEquals:
aws:SourceArn: !Ref UserEventsTopic3. 順序保証型：イベントの順序が重要な場合
これは地味に難しいんですよ。SNSはデフォルトでは順序を保証しません。ユーザーが「作成」→「更新」→「削除」の順で実行されるはずが、順番がバラバラに到達することもあります。
こういう場合は、SQS FIFO キューを使うのが正解。でも、SNSはFIFOトピックを使うと、複数キューへの配信時に順序が保証される特性があります。
# FIFO構成での発行
publisher = EventPublisher(topic_arn='arn:aws:sns:ap-northeast-1:xxx:user-events.fifo')
# FIFO送信には、MessageGroupIdが必須
sns.publish(
TopicArn='arn:aws:sns:ap-northeast-1:xxx:user-events.fifo',
Message=json.dumps(event_data),
MessageGroupId=f'user-{user_id}', # 同じユーザーなら同じグループ
MessageDeduplicationId=f'{user_id}-{event_timestamp}' # 重複排除
)ただ、FIFOを使うとスループットが10倍以上落ちるんです。秒あたり300メッセージが上限。正直、本番で「本当に順序保証が必要か？」を厳密に検証してから導入すべき。
DLQ設計：地雷の宝庫
DLQ（Dead Letter Queue）の設定で、うちのチームが失敗したのは以下の3つですね。
1. DLQへの転送タイミングの誤解
最初は「エラーが発生したら即DLQへ」と思ってたんですが、これだと一時的なネットワーク障害とか、一過性のエラーまで全部DLQに送られちゃう。改善策は「ApproximateReceiveCount」を監視して、複数回失敗したときだけDLQへ送ることです。上のコード例で実装してます。
2. DLQ監視の放置
DLQに送られたメッセージって、放置されてることが多いんですよ。週1回はDLQを確認して、なぜ失敗したのかログを見るべき。CloudWatch Alarmで「DLQ に新しいメッセージが入った」というアラートを設定するのが重要です。
3. リテンション期間のズレ
メインキューのリテンション期間が14日で、DLQが4日だと、メインキューから削除される前にDLQの調査期限が終わっちゃう。両方を揃える、できれば同じ期間にするべきです。
スループット最適化：本番で見えた課題
SQS/SNS連携でスループットを上げるには、いくつかコツがあります。
# スループット最適化版
class HighThroughputConsumer:
def __init__(self, queue_url, num_workers=10):
self.queue_url = queue_url
self.num_workers = num_workers
def consume_parallel(self, handler_func):
"""複数ワーカーで並列処理"""
with ThreadPoolExecutor(max_workers=self.num_workers) as executor:
futures = []
while True:
# バッチで受信（最大10メッセージ）
response = sqs.receive_message(
QueueUrl=self.queue_url,
MaxNumberOfMessages=10,
WaitTimeSeconds=20
)
if 'Messages' not in response:
time.sleep(0.5)
continue
for message in response['Messages']:
# 非同期で処理
future = executor.submit(
self._handle_message,
message,
handler_func
)
futures.append(future)
# 完了を待たずに次のバッチを取得（パイプライン化）
while futures:
done = [f for f in futures if f.done()]
for f in done:
try:
f.result(timeout=0.1)
except Exception as e:
logger.error(f"Task failed: {e}")
futures.remove(f)
if len(futures) < self.num_workers:
break
def _handle_message(self, message, handler_func):
try:
body = json.loads(message['Body'])
handler_func(body)
sqs.delete_message(
QueueUrl=self.queue_url,
ReceiptHandle=message['ReceiptHandle']
)
except Exception as e:
logger.error(f"Error: {e}")
raise重要なポイントをまとめると以下の通り。
| 項目 | 推奨値 | 理由 |
|---|---|---|
| BatchSize | 10 | SQSの最大値。リクエスト削減 |
| WaitTimeSeconds | 20 | Long Polling。無駄なリクエスト削減 |
| ワーカー数 | 10～20 | 並列処理でスループット向上 |
| VisibilityTimeout | 処理時間×1.5 | 重複処理を防ぐ |
AWS構成図：本番で安定する設計
flowchart TB
subgraph Application[" Application Layer"]
API["API Gateway<br/>Lambda"]
end
subgraph EventBridge["Event Distribution"]
SNS["SNS Topic<br/>user-events"]
end
subgraph HighPriority["High Priority Path"]
subgraph HP["High Priority Queue"]
HPQUEUE["SQS Queue<br/>Premium Users"]
HPDLQ["DLQ<br/>max_retries=3"]
end
HPWORKER["Lambda Consumer<br/>Workers=20<br/>Timeout=300s"]
end
subgraph LowPriority["Low Priority Path"]
subgraph LP["Low Priority Queue"]
LPQUEUE["SQS Queue<br/>Free Users"]
LPDLQ["DLQ<br/>max_retries=5"]
end
LPWORKER["Lambda Consumer<br/>Workers=5<br/>Timeout=600s"]
end
subgraph Monitoring["Monitoring & Alerts"]
CLOUDWATCH["CloudWatch<br/>Metrics"]
ALARM["SNS Alarm<br/>DLQ Messages > 10"]
end
subgraph DataLake["Data Persistence"]
DYNAMODB["DynamoDB<br/>Event Log"]
S3["S3<br/>Archive<br/>7 days"]
end
API -->|publish event| SNS
SNS -->|filter by user_tier| HPQUEUE
SNS -->|filter by user_tier| LPQUEUE
HPQUEUE -->|consume<br/>batch=10| HPWORKER
HPWORKER -->|success| DYNAMODB
HPWORKER -->|fail 3x| HPDLQ
HPDLQ -->|alert| ALARM
LPQUEUE -->|consume<br/>batch=10| LPWORKER
LPWORKER -->|success| DYNAMODB
LPWORKER -->|fail 5x| LPDLQ
LPDLQ -->|alert| ALARM
HPQUEUE -->|metrics| CLOUDWATCH
LPQUEUE -->|metrics| CLOUDWATCH
HPDLQ -->|metrics| CLOUDWATCH
LPDLQ -->|metrics| CLOUDWATCH
DYNAMODB -->|archive| S3
ALARM -->|notify| Application実際に本番で起きた問題と対策
問題1：メッセージ重複
SNS→SQS変換時に、同じメッセージが複数回キューに入ることがあります。これは「at least once」の配信保証が理由。対策は、メッセージにユニークなIDを付与して、消費側で重複排除することです。
# メッセージIDによる重複排除
processed_ids = set()
def handle_with_dedup(event):
message_id = event.get('message_id')
# DynamoDBで見た履歴があるか確認
try:
dynamodb.get_item(
TableName='processed-messages',
Key={'message_id': {'S': message_id}}
)
logger.info(f"Message already processed: {message_id}")
return # スキップ
except:
pass
# 処理実行
process_event(event)
# 処理済みとして記録
dynamodb.put_item(
TableName='processed-messages',
Item={
'message_id': {'S': message_id},
'timestamp': {'N': str(int(time.time()))}
},
TimeToLive=int(time.time()) + 86400 # 24時間後に自動削除
)問題2：可視性タイムアウトの設定ミス
VisibilityTimeout が処理時間より短いと、処理中にメッセージが再度キューに現れて、別のワーカーが同じメッセージを拾う。結果、重複処理やデータ不整合が起きます。
対策は、処理時間 + 余裕 を設定すること。また、処理中に「あと30秒必要」と気づいたら、change_message_visibility で延長するのが良いです。
# 処理中にタイムアウトを延長
def process_with_visibility_extension(queue_url, receipt_handle, handler):
try:
# 重い処理の途中
result = handler() # 長時間かかる可能性
# あと1分必要と判断
sqs.change_message_visibility(
QueueUrl=queue_url,
ReceiptHandle=receipt_handle,
VisibilityTimeout=60
)
# 継続処理
finalize(result)
except Exception as e:
logger.error(f"Processing failed: {e}")
raise問題3：メッセージサイズの制限
SQS のメッセージサイズは256KBが上限。大きなペイロードを送りたいときは、S3に保存してURLだけをSQSに送るポインター方式を使います。
import uuid
def publish_large_event(event_data):
payload = json.dumps(event_data)
if len(payload.encode()) > 256 * 1024: # 256KB超過
# S3に保存
key = f"events/{uuid.uuid4()}.json"
s3.put_object(
Bucket='event-storage',
Key=key,
Body=payload,
ServerSideEncryption='AES256'
)
# SQSには参照情報だけ送信
sns.publish(
TopicArn='arn:aws:sns:...',
Message=json.dumps({
's3_bucket': 'event-storage',
's3_key': key,
'event_type': event_data.get('type'),
'timestamp': int(time.time())
})
)
else:
# 通常送信
sns.publish(
TopicArn='arn:aws:sns:...',
Message=payload
)チームで整理して気づいた、本当に使い分けるべきポイント
実は、SNS/SQS以外の選択肢も視野に入れるべきです。うちのチームでも「Kafkaではなくて本当にSQS/SNSで良いのか？」という議論を何度もしました。
| 要件 | SQS/SNS向き | Kafka向き |
|---|---|---|
| イベントの種類 | 5～10個程度 | 大量 |
| 順序保証 | 不要 | 必須 |
| スループット | 1000msg/秒以下 | 10000msg/秒以上 |
| リプレイ機能 | 不要 | 必須 |
| 費用最適化 | 重視 | 重視不要 |
正直、「どちらを選ぶか」は最初の検討が大事。後から「Kafkaに乗り換えたい」となると、マイグレーションが地獄になります。
まとめ
SQS/SNS連携は見た目よりずっと奥が深い。本番で学んだ重要なポイントを最後にまとめますね。
- Fan-outは各サービスに独立したキューを用意する——スループットの競合を避けるため
- DLQの監視は放置するな——週1回は確認して、失敗パターンを分析する
- VisibilityTimeoutと処理時間の関係を厳密に計算する——重複処理の大半はここから生まれる
- メッセージ重複排除は必須——at least once配信は避けられないから
- 本当に必要な機能か冷静に判断する——FIFO、優先度分け、順序保証は全部コスト増
次のアクション、自分たちのシステムで「今使ってるSNS/SQS設定」を見直すチャンスです。もし「FIFOを使ってるけど、実は順序不要」「DLQのアラームを設定してない」みたいなことがあれば、改善の余地あり。正直、本番で何度も同じ失敗するより、設計段階で気づいておく方が絶対に得です。
関連記事
サーバーレス設計