メッセージキューの設計パターン——非同期処理、イベント駆動、信頼性のある配信を実現する

import boto3
import json

# AWS SQS のクライアント
sqs = boto3.client('sqs')

# キュー URL の取得
queue_url = sqs.get_queue_url(QueueName='my-queue')['QueueUrl']

# メッセージ送信
response = sqs.send_message(
    QueueUrl=queue_url,
    MessageBody=json.dumps({
        "type": "email_send",
        "to": "user@example.com",
        "subject": "Welcome!",
        "body": "Thank you for signing up."
    }),
    MessageAttributes={
        "MessageType": {
            "DataType": "String",
            "StringValue": "Notification"
        }
    }
)

print(f"メッセージ ID: {response['MessageId']}")

# 受信側（複数ワーカーで競合処理）
def worker():
    while True:
        response = sqs.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=1,
            WaitTimeSeconds=10  # ロングポーリング
        )

        if 'Messages' not in response:
            continue

        message = response['Messages'][0]
        body = json.loads(message['Body'])

        try:
            process_task(body)
            # 処理成功：キューから削除
            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=message['ReceiptHandle']
            )
        except Exception as e:
            # 処理失敗：可視化タイマーが切れてキューに戻る
            print(f"処理失敗：{e}")

# SNS（パブ・サブ）での送信
sns = boto3.client('sns')
topic_arn = "arn:aws:sns:us-east-1:123456789012:my-topic"

# トピックに送信（すべての購読者に配信される）
response = sns.publish(
    TopicArn=topic_arn,
    Message=json.dumps({
        "event": "user.created",
        "user_id": "usr_123",
        "email": "user@example.com"
    }),
    Subject="新規ユーザー登録"
)

print(f"メッセージ ID: {response['MessageId']}")

import pika

# RabbitMQ 接続
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Exchange 宣言
channel.exchange_declare(exchange='logs', exchange_type='topic')

# パターンマッチでルーティング
# app.email.error → error キューと email キュー両方に
channel.queue_bind(exchange='logs', queue='error', routing_key='*.error')
channel.queue_bind(exchange='logs', queue='email', routing_key='email.*')

# 送信
channel.basic_publish(
    exchange='logs',
    routing_key='app.email.info',
    body=json.dumps({"msg": "Email sent"})
)

from kafka import KafkaProducer, KafkaConsumer
import json

# Producer
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# メッセージ送信
producer.send('user-events', {
    "event": "user.created",
    "user_id": "usr_123"
})
producer.flush()

# Consumer
consumer = KafkaConsumer(
    'user-events',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    group_id='analytics-service',
    auto_offset_reset='earliest'
)

for message in consumer:
    print(f"受信：{message.value}")

import time
import random

def process_with_retry(message, max_retries=3):
    """指数バックオフ付きリトライ"""
    for attempt in range(max_retries):
        try:
            return process_message(message)
        except TemporaryError as e:
            if attempt == max_retries - 1:
                raise  # 最大リトライ回数超え

            # 指数バックオフ（jitter 付き）
            base_delay = 2 ** attempt  # 2, 4, 8 秒
            jitter = random.uniform(0, base_delay * 0.1)
            delay = base_delay + jitter

            print(f"リトライ {attempt + 1}/{max_retries}、{delay:.1f}秒後に再試行")
            time.sleep(delay)

def process_message(message):
    """実際の処理（一時的なエラーが発生する可能性）"""
    if random.random() < 0.3:  # 30% の確率で失敗
        raise TemporaryError("一時的なエラー")
    print(f"処理成功：{message}")

# AWS SQS の DLQ 設定
import boto3

sqs = boto3.client('sqs')

# DLQ 用キューを作成
dlq_response = sqs.create_queue(QueueName='my-queue-dlq')
dlq_url = dlq_response['QueueUrl']

# DLQ の ARN を取得
dlq_arn = sqs.get_queue_attributes(
    QueueUrl=dlq_url,
    AttributeNames=['QueueArn']
)['Attributes']['QueueArn']

# メインキューに DLQ 設定を紐付ける
redrive_policy = {
    "deadLetterTargetArn": dlq_arn,
    "maxReceiveCount": "3"  # 3 回失敗すると DLQ 行き
}

main_response = sqs.create_queue(
    QueueName='my-queue',
    Attributes={
        'RedrivePolicy': json.dumps(redrive_policy),
        'VisibilityTimeout': '300'  # 5 分
    }
)

# DLQ のメッセージを確認
dlq_messages = sqs.receive_message(
    QueueUrl=dlq_url,
    MaxNumberOfMessages=10,
    WaitTimeSeconds=5
)

if 'Messages' in dlq_messages:
    print(f"DLQ に {len(dlq_messages['Messages'])} 件の失敗メッセージ")

    for msg in dlq_messages['Messages']:
        body = json.loads(msg['Body'])
        print(f"失敗内容：{body}")
        # 必要に応じて再処理
        # sqs.send_message(QueueUrl=main_queue_url, MessageBody=msg['Body'])

# FIFO キューへの送信（末尾に .fifo が必要）
fifo_queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue.fifo"

# ユーザーごとの順序保証
sqs.send_message(
    QueueUrl=fifo_queue_url,
    MessageBody=json.dumps({"action": "deposit", "amount": 10000}),
    MessageGroupId="user_123",  # 同一ユーザーは順番に処理
    MessageDeduplicationId="msg_001"
)

sqs.send_message(
    QueueUrl=fifo_queue_url,
    MessageBody=json.dumps({"action": "withdraw", "amount": 5000}),
    MessageGroupId="user_123",  # 入金後に処理
    MessageDeduplicationId="msg_002"
)

# ユーザー ID をキーとして使用（同一ユーザーは常に同一パーティションへ）
producer.send(
    'order-events',
    key=b"user_123",  # 同一キーは同一パーティション
    value={
        "order_id": "ord_001",
        "status": "created"
    }
)

# イベントストア
events = [
    {"type": "OrderCreated", "order_id": "ord_001", "timestamp": "2025-01-01T10:00:00Z"},
    {"type": "OrderPaid", "order_id": "ord_001", "timestamp": "2025-01-01T10:05:00Z"},
    {"type": "OrderShipped", "order_id": "ord_001", "timestamp": "2025-01-01T15:00:00Z"},
]

# 現在の状態をイベントから再構築
def reconstruct_order(events):
    order = None
    for event in events:
        if event["type"] == "OrderCreated":
            order = {"id": event["order_id"], "status": "created"}
        elif event["type"] == "OrderPaid":
            order["status"] = "paid"
        elif event["type"] == "OrderShipped":
            order["status"] = "shipped"
    return order

order = reconstruct_order(events)
# {"id": "ord_001", "status": "shipped"}

# コマンド側（書き込み）
def create_order(command):
    # 書き込み DB に保存
    order = db.orders.insert({
        "id": generate_id(),
        "user_id": command["user_id"],
        "status": "created"
    })

    # イベント発行
    event_bus.publish({
        "type": "OrderCreated",
        "order_id": order["id"]
    })

# クエリ側（読み込み）
def get_order_summary(user_id):
    # 読み込み専用ビュー（キャッシュ済み）
    return redis.get(f"order_summary:{user_id}")

def place_order_saga(order_data):
    """Saga による分散トランザクション"""
    saga_id = generate_saga_id()
    compensations = []  # 補償トランザクション

    try:
        # ステップ 1: 在庫確保
        inventory_result = inventory_service.reserve(order_data)
        compensations.append(lambda: inventory_service.release(inventory_result))

        # ステップ 2: 決済実行
        payment_result = payment_service.charge(order_data)
        compensations.append(lambda: payment_service.refund(payment_result))

        # ステップ 3: 注文確定
        order_service.confirm(order_data)

        return {"status": "success"}

    except Exception as e:
        # 失敗：補償トランザクションを逆順で実行
        print(f"Saga 失敗、補償を実行：{e}")
        for compensation in reversed(compensations):
            try:
                compensation()
            except Exception as comp_error:
                print(f"補償失敗：{comp_error}")
                # 手動介入が必要な場合も

        return {"status": "failed", "reason": str(e)}

# 重複チェック付き処理
def process_message_idempotent(message_id, data):
    # Redis で既処理チェック
    if redis.exists(f"processed:{message_id}"):
        print(f"重複メッセージ：{message_id}")
        return

    # 処理実行
    process_data(data)

    # 処理済みマーク（TTL 付き）
    redis.setex(f"processed:{message_id}", 86400, "1")  # 24 時間

def process_with_circuit_breaker(message):
    """サーキットブレーカー付き処理"""
    error_count = redis.get(f"error_count:{message['type']}") or 0

    if int(error_count) > 10:
        # 10 回連続失敗：サーキットオープン、DLQ 行き
        send_to_dlq(message, reason="circuit_open")
        return

    try:
        process_message(message)
        redis.delete(f"error_count:{message['type']}")  # カウントリセット
    except Exception as e:
        redis.incr(f"error_count:{message['type']}")
        redis.expire(f"error_count:{message['type']}", 300)  # 5 分でリセット
        raise

# キュー深度の監視
def check_queue_depth(queue_url):
    attrs = sqs.get_queue_attributes(
        QueueUrl=queue_url,
        AttributeNames=['ApproximateNumberOfMessages']
    )
    depth = int(attrs['Attributes']['ApproximateNumberOfMessages'])

    if depth > 10000:
        # 閾値超え：アラート + 自動スケーリング
        alert_slack(f"キュー深度が閾値超え：{depth}")
        trigger_auto_scaling()

    return depth