メッセージキューの基礎——非同期処理とイベント駆動アーキテクチャの基盤

class MessageQueue:
    """メッセージキューの基本インターフェース"""

    def publish(self, queue_name: str, message: dict) -> str:
        """
        メッセージをキューに送信

        Args:
            queue_name: 宛先キュー名
            message: 送信メッセージ（辞書形式）

        Returns:
            message_id: 割り当てられたメッセージ ID
        """
        raise NotImplementedError

    def consume(self, queue_name: str, timeout: int = 30) -> dict | None:
        """
        メッセージをキューから取得

        Args:
            queue_name: 対象キュー名
            timeout: タイムアウト秒数

        Returns:
            message: 取得したメッセージ、または None（タイムアウト時）
        """
        raise NotImplementedError

    def acknowledge(self, message_id: str) -> bool:
        """
        メッセージ処理完了の承認

        Args:
            message_id: 処理完了したメッセージ ID

        Returns:
            success: 承認成功 True
        """
        raise NotImplementedError

    def nack(self, message_id: str, requeue: bool = True) -> bool:
        """
        メッセージ処理失敗の通知

        Args:
            message_id: 処理失敗したメッセージ ID
            requeue: True ならキューに再投入

        Returns:
            success: 通知成功 True
        """
        raise NotImplementedError

def create_message(body: dict, properties: dict | None = None) -> dict:
    """
    メッセージオブジェクトの生成

    Args:
        body: メッセージ本文（ビジネスデータ）
        properties: プロパティ（メタデータ）

    Returns:
        message: 完全なメッセージオブジェクト
    """
    import uuid
    from datetime import datetime

    return {
        "message_id": str(uuid.uuid4()),
        "timestamp": datetime.utcnow().isoformat(),
        "body": body,
        "properties": properties or {},
        "headers": {
            "content_type": "application/json",
            "delivery_mode": 2,  # 永続化
        }
    }

# 使用例
order_message = create_message(
    body={
        "order_id": "ORD-2025-001",
        "customer_id": "CUST-123",
        "items": [
            {"product_id": "PROD-A", "quantity": 2},
            {"product_id": "PROD-B", "quantity": 1}
        ],
        "total_amount": 5000
    },
    properties={
        "correlation_id": "CORR-456",
        "reply_to": "response-queue-001"
    }
)

print(f"メッセージ ID: {order_message['message_id']}")
print(f"本文：{order_message['body']}")

def analyze_delivery_guarantees():
    """
    配信保証の比較分析
    """
    guarantees = {
        "at_most_once": {
            "description": "最大 1 回配信（損失の可能性あり）",
            "mechanism": "ack なし",
            "use_case": "ログ収集、メトリクス（欠落許容）",
            "performance": "最高"
        },
        "at_least_once": {
            "description": "最小 1 回配信（重複の可能性あり）",
            "mechanism": "ack + 再配信",
            "use_case": "注文処理、決済（欠落不允许）",
            "performance": "中"
        },
        "exactly_once": {
            "description": "丁度 1 回配信（理想的）",
            "mechanism": "べき性 + トランザクション",
            "use_case": "残高更新、重複不允许",
            "performance": "最低"
        }
    }

    return guarantees

delivery_analysis = analyze_delivery_guarantees()

for key, value in delivery_analysis.items():
    print(f"\n{key}:")
    for k, v in value.items():
        print(f"  {k}: {v}")

class PointToPointQueue:
    """Point-to-Point パターンの実装例"""

    def __init__(self, queue_name: str):
        self.queue_name = queue_name
        self.consumers = []  # 競合コンシューマー群

    def register_consumer(self, consumer_func):
        """コンシューマーの登録"""
        self.consumers.append(consumer_func)

    def dispatch(self, message):
        """
        メッセージの配信（1 つのコンシューマーにのみ）
        """
        if not self.consumers:
            return False

        # ラウンドロビンで配信
        consumer = self.consumers[len(self.consumers) % len(self.consumers)]
        consumer(message)
        return True

# 使用例
order_queue = PointToPointQueue("order-queue")

def consumer_a(message):
    print(f"Consumer A: 処理 {message['order_id']}")

def consumer_b(message):
    print(f"Consumer B: 処理 {message['order_id']}")

order_queue.register_consumer(consumer_a)
order_queue.register_consumer(consumer_b)

# 負荷分散：A → B → A → B ...

class PubSubBroker:
    """Publish-Subscribe パターンの実装例"""

    def __init__(self):
        self.subscriptions = {}  # topic -> [subscribers]

    def subscribe(self, topic: str, callback):
        """
        トピックの購読登録

        Args:
            topic: 購読トピック
            callback: メッセージ受信時のコールバック
        """
        if topic not in self.subscriptions:
            self.subscriptions[topic] = []
        self.subscriptions[topic].append(callback)
        print(f"購読開始：{topic}")

    def unsubscribe(self, topic: str, callback):
        """購読解除"""
        if topic in self.subscriptions:
            self.subscriptions[topic].remove(callback)

    def publish(self, topic: str, message: dict):
        """
        トピックへのメッセージ配信

        Args:
            topic: 配信トピック
            message: 配信メッセージ
        """
        if topic in self.subscriptions:
            for callback in self.subscriptions[topic]:
                callback(message)
            print(f"配信完了：{topic} -> {len(self.subscriptions[topic])} 購読者")

# 使用例
broker = PubSubBroker()

# 注文イベントの購読者
def inventory_service(event):
    print(f"在庫サービス：在庫更新 {event['order_id']}")

def shipping_service(event):
    print(f"配送サービス：配送準備 {event['order_id']}")

def analytics_service(event):
    print(f"分析サービス：ログ記録 {event['order_id']}")

broker.subscribe("order.created", inventory_service)
broker.subscribe("order.created", shipping_service)
broker.subscribe("order.created", analytics_service)

# 注文確定イベントを配信
broker.publish("order.created", {
    "order_id": "ORD-2025-001",
    "customer_id": "CUST-123"
})

import threading
import queue
import time

class WorkerPool:
    """
    ワーカープールの実装例
    """

    def __init__(self, num_workers: int = 4):
        self.task_queue = queue.Queue()
        self.workers = []
        self.num_workers = num_workers

    def start(self):
        """ワーカー起動"""
        for i in range(self.num_workers):
            worker = threading.Thread(
                target=self._worker_loop,
                name=f"Worker-{i}",
                daemon=True
            )
            worker.start()
            self.workers.append(worker)
        print(f"ワーカープール起動：{self.num_workers} ワーカー")

    def _worker_loop(self):
        """ワーカー処理ループ"""
        thread_name = threading.current_thread().name
        while True:
            try:
                task = self.task_queue.get(timeout=1)
                if task is None:  # ポイズンピル
                    break

                print(f"[{thread_name}] タスク処理開始：{task['id']}")
                task['func'](*task['args'], **task['kwargs'])
                print(f"[{thread_name}] タスク処理完了：{task['id']}")
                self.task_queue.task_done()
            except queue.Empty:
                continue

    def submit(self, func, *args, **kwargs):
        """
        タスクの投入

        Args:
            func: 実行関数
            *args: 位置引数
            **kwargs: キーワード引数
        """
        task = {
            "id": f"task-{time.time()}",
            "func": func,
            "args": args,
            "kwargs": kwargs
        }
        self.task_queue.put(task)
        print(f"タスク投入：{task['id']}")

# 使用例
def process_email(to, subject, body):
    """メール送信タスク（時間がかかる処理）"""
    time.sleep(0.5)  # 模擬遅延
    print(f"メール送信：{to} - {subject}")

pool = WorkerPool(num_workers=3)
pool.start()

# 100 通のメール送信タスクを投入
for i in range(100):
    pool.submit(process_email, f"user{i}@example.com", "ようこそ", "会員登録ありがとうございます")

# 全タスクの完了を待機
pool.task_queue.join()
print("全タスク完了")

from collections import deque
from dataclasses import dataclass
from datetime import datetime

@dataclass
class MessageWithRetry:
    message: dict
    retry_count: int = 0
    max_retries: int = 3
    last_error: str | None = None

class DeadLetterQueue:
    """デッドレターキューの実装例"""

    def __init__(self, main_queue_name: str, dlq_name: str, max_retries: int = 3):
        self.main_queue = deque()
        self.dlq = deque()
        self.main_queue_name = main_queue_name
        self.dlq_name = dlq_name
        self.max_retries = max_retries

    def send_to_main(self, message: dict):
        """メインキューへ送信"""
        msg = MessageWithRetry(message=message, max_retries=self.max_retries)
        self.main_queue.append(msg)
        print(f"メインキューへ：{message.get('id', 'unknown')}")

    def process_message(self) -> bool:
        """
        メインキューの処理

        Returns:
            success: True if processed successfully
        """
        if not self.main_queue:
            return False

        msg = self.main_queue.popleft()

        try:
            # 模擬処理（50% の確率で失敗）
            import random
            if random.random() < 0.5:
                raise Exception("模擬エラー")

            print(f"処理成功：{msg.message.get('id')}")
            return True

        except Exception as e:
            msg.retry_count += 1
            msg.last_error = str(e)

            if msg.retry_count < msg.max_retries:
                print(f"リトライ：{msg.message.get('id')} ({msg.retry_count}/{msg.max_retries})")
                self.main_queue.append(msg)  # 再投入
            else:
                print(f"DLQ へ移動：{msg.message.get('id')} (最大リトライ回数超過)")
                self.dlq.append(msg)
            return False

    def get_dlq_messages(self):
        """DLQ のメッセージ一覧取得"""
        return list(self.dlq)

# 使用例
dlq_handler = DeadLetterQueue("main-queue", "dead-letter-queue", max_retries=3)

# テストメッセージを投入
for i in range(5):
    dlq_handler.send_to_main({"id": f"msg-{i}", "data": f"test-{i}"})

# 処理実行（リトライと DLQ 移動を確認）
for _ in range(10):
    dlq_handler.process_message()

# DLQ の内容確認
print(f"\nDLQ 待機メッセージ：{len(dlq_handler.get_dlq_messages())} 件")

def select_message_queue(requirements: dict) -> str:
    """
    要件に基づくメッセージキューの選定

    Args:
        requirements: 要件定義（辞書形式）

    Returns:
        recommended: 推奨メッセージキュー
    """
    score = {
        "RabbitMQ": 0,
        "Kafka": 0,
        "AWS SQS": 0,
        "Redis Streams": 0
    }

    # 高スループットが必要
    if requirements.get("high_throughput"):
        score["Kafka"] += 3
        score["Redis Streams"] += 3
        score["RabbitMQ"] += 1

    # 低レイテンシが必要
    if requirements.get("low_latency"):
        score["Redis Streams"] += 3
        score["RabbitMQ"] += 2

    # 複雑なルーティングが必要
    if requirements.get("complex_routing"):
        score["RabbitMQ"] += 3
        score["Kafka"] += 1

    # 長期保存が必要
    if requirements.get("long_retention"):
        score["Kafka"] += 3
        score["AWS SQS"] += 2

    # 運用コストを低くしたい
    if requirements.get("low_ops_cost"):
        score["AWS SQS"] += 3
        score["Redis Streams"] += 1

    # イベントソーシング
    if requirements.get("event_sourcing"):
        score["Kafka"] += 3

    # マルチキャスト/プッシュ配信
    if requirements.get("multicast"):
        score["RabbitMQ"] += 3
        score["Kafka"] += 2

    # 最大スコアの製品を推奨
    return max(score, key=score.get)

# 使用例：EC サイトの注文処理システム
requirements_ec = {
    "high_throughput": False,  # 1 秒あたり数百件
    "low_latency": True,  # 応答性重視
    "complex_routing": True,  # 在庫、配送、通知へルーティング
    "long_retention": False,  # 消費すれば不要
    "low_ops_cost": False,  # 自前運用可
    "event_sourcing": False,
    "multicast": True  # 複数サービスへ通知
}

recommended = select_message_queue(requirements_ec)
print(f"EC サイト注文処理：{recommended} が推奨")  # RabbitMQ

# 使用例：ログ収集・分析システム
requirements_log = {
    "high_throughput": True,  # 1 秒あたり数万イベント
    "low_latency": False,  # 数秒の遅延は許容
    "complex_routing": False,  # 単純なパイプライン
    "long_retention": True,  # 7 日間保存
    "low_ops_cost": False,
    "event_sourcing": True,  # イベントログの蓄積
    "multicast": False
}

recommended_log = select_message_queue(requirements_log)
print(f"ログ収集システム：{recommended_log} が推奨")  # Kafka

def calculate_optimal_settings(throughput_target: int, avg_message_size: int):
    """
    最適設定の試算

    Args:
        throughput_target: 目標スループット（msg/s）
        avg_message_size: 平均メッセージサイズ（バイト）

    Returns:
        settings: 推奨設定
    """
    # 必要なバッチサイズ
    batch_size = min(throughput_target // 100, 1000)  # 最大 1000

    # プリフェッチカウント（一度に取得するメッセージ数）
    prefetch_count = batch_size // 10

    # 必要なパーティション数（Kafka）
    partitions = max(1, throughput_target // 1000)

    return {
        "batch_size": batch_size,
        "prefetch_count": prefetch_count,
        "partitions": partitions,
        "consumer_parallelism": partitions * 2
    }

# 使用例：1 秒あたり 5000 メッセージの処理
settings = calculate_optimal_settings(5000, 1024)

print("最適設定（目標：5000 msg/s）:")
for key, value in settings.items():
    print(f"  {key}: {value}")

from typing import Any
import hashlib

class IdempotentProcessor:
    """
    べき性のあるメッセージ処理

    同じメッセージ ID に対する処理が 2 回実行されないことを保証
    """

    def __init__(self):
        self.processed_ids = set()

    def compute_message_hash(self, message: dict) -> str:
        """メッセージのハッシュ値計算（重複検出用）"""
        content = f"{message.get('id')}:{message.get('timestamp')}:{message.get('body')}"
        return hashlib.sha256(content.encode()).hexdigest()

    def process(self, message: dict) -> dict | None:
        """
        メッセージ処理（べき性保証）

        Args:
            message: 処理対象メッセージ

        Returns:
            result: 処理結果（重複の場合は None）
        """
        msg_hash = self.compute_message_hash(message)

        # 重複チェック
        if msg_hash in self.processed_ids:
            print(f"重複メッセージ：スキップ {message.get('id')}")
            return None

        # 実際の処理
        result = self._do_process(message)

        # 処理済みマーク
        self.processed_ids.add(msg_hash)

        return result

    def _do_process(self, message: dict) -> dict:
        """実際の処理ロジック"""
        # 例：注文処理
        order = message.get('body', {})
        return {
            "status": "processed",
            "order_id": order.get('order_id'),
            "amount": order.get('total_amount')
        }

# 使用例：at_least_once 配信での重複防止
processor = IdempotentProcessor()

# 1 回目の処理
message = {
    "id": "msg-001",
    "timestamp": "2025-09-05T10:00:00Z",
    "body": {"order_id": "ORD-001", "total_amount": 5000}
}

result1 = processor.process(message)
print(f"1 回目：{result1}")  # 処理される

# 2 回目の処理（ネットワーク障害で再配信）
result2 = processor.process(message)
print(f"2 回目：{result2}")  # None（スキップ）

from contextlib import contextmanager

class TransactionalProcessor:
    """
    トランザクションメッセージ処理

    データベース更新とメッセージ ack を アトミックに実行
    """

    def __init__(self, db_connection, message_queue):
        self.db = db_connection
        self.mq = message_queue

    @contextmanager
    def transaction(self):
        """
        トランザクションコンテキストマネージャ

        使用例:
        with processor.transaction():
            processor.db_update(data)
            processor.mq.ack(message_id)
        """
        try:
            self.db.begin()
            yield
            self.db.commit()
        except Exception as e:
            self.db.rollback()
            raise e

    def process_order(self, message: dict):
        """
        注文処理（トランザクション）
        """
        with self.transaction():
            # データベース更新
            order = message['body']
            self.db.execute(
                "INSERT INTO orders (id, customer_id, amount) VALUES (?, ?, ?)",
                order['order_id'], order['customer_id'], order['total_amount']
            )

            # 在庫更新
            for item in order['items']:
                self.db.execute(
                    "UPDATE inventory SET quantity = quantity - ? WHERE product_id = ?",
                    item['quantity'], item['product_id']
                )

            # メッセージ ack
            self.mq.acknowledge(message['message_id'])

            print(f"注文処理完了：{order['order_id']}")

# トランザクションの重要性：
# - メッセージ ack だけ成功して DB 更新失敗 → データ消失
# - DB 更新だけ成功して ack 失敗 → 重複処理
# → 両方をアトミックに実行する必要がある

import json
import pickle
from dataclasses import dataclass, asdict

@dataclass
class Order:
    order_id: str
    customer_id: str
    items: list
    total_amount: int

class MessageSerializer:
    """メッセージのシリアライゼーション比較"""

    @staticmethod
    def json_serialize(order: Order) -> bytes:
        """JSON シリアライゼーション"""
        return json.dumps(asdict(order)).encode('utf-8')

    @staticmethod
    def json_deserialize(data: bytes) -> Order:
        """JSON デシリアライゼーション"""
        data_dict = json.loads(data.decode('utf-8'))
        return Order(**data_dict)

    @staticmethod
    def pickle_serialize(order: Order) -> bytes:
        """Pickle シリアライゼーション"""
        return pickle.dumps(order)

    @staticmethod
    def pickle_deserialize(data: bytes) -> Order:
        """Pickle デシリアライゼーション"""
        return pickle.loads(data)

# 使用例
order = Order(
    order_id="ORD-2025-001",
    customer_id="CUST-123",
    items=[{"product_id": "PROD-A", "quantity": 2}],
    total_amount=5000
)

# JSON（推奨：言語間互換性あり）
json_data = MessageSerializer.json_serialize(order)
json_restored = MessageSerializer.json_deserialize(json_data)
print(f"JSON サイズ：{len(json_data)} bytes")

# Pickle（Python 専用：セキュリティリスクあり）
pickle_data = MessageSerializer.pickle_serialize(order)
pickle_restored = MessageSerializer.pickle_deserialize(pickle_data)
print(f"Pickle サイズ：{len(pickle_data)} bytes")

# 比較:
# JSON: 可読性○、言語間互換性○、セキュリティ○
# Pickle: 可読性×、言語間互換性×、セキュリティ△（不正データ実行リスク）

from typing import Literal

class MessageVersioning:
    """
    メッセージのバージョニング戦略

    スキーマ進化に対応するためのバージョン管理
    """

    def __init__(self):
        self.schema_registry = {}

    def register_schema(self, message_type: str, version: int, schema: dict):
        """スキーマの登録"""
        key = f"{message_type}:v{version}"
        self.schema_registry[key] = schema
        print(f"スキーマ登録：{key}")

    def validate(self, message_type: str, version: int, message: dict) -> bool:
        """
        メッセージのスキーマ検証

        Args:
            message_type: メッセージタイプ
            version: スキーマバージョン
            message: 検証対象メッセージ

        Returns:
            valid: True if valid
        """
        key = f"{message_type}:v{version}"
        if key not in self.schema_registry:
            print(f"未知のスキーマ：{key}")
            return False

        schema = self.schema_registry[key]
        required_fields = schema.get('required', [])

        for field in required_fields:
            if field not in message:
                print(f"必須フィールド不足：{field}")
                return False

        return True

    def migrate(self, message: dict, from_version: int, to_version: int) -> dict:
        """
        メッセージのバージョンマイグレーション

        Args:
            message: 移行元メッセージ
            from_version: 移行元バージョン
            to_version: 移行先バージョン

        Returns:
            migrated: 移行後メッセージ
        """
        migrated = message.copy()
        migrated['version'] = to_version

        # 単純な移行例：v1 → v2（customer_id を customer 情報に拡張）
        if from_version == 1 and to_version == 2:
            migrated['customer'] = {
                'id': migrated.pop('customer_id'),
                'name': 'Unknown',  # デフォルト値
                'email': None
            }

        print(f"バージョン移行：v{from_version} → v{to_version}")
        return migrated

# 使用例
versioning = MessageVersioning()

# v1 スキーマ登録
versioning.register_schema("order", 1, {
    "type": "object",
    "required": ["order_id", "customer_id", "items"]
})

# v2 スキーマ登録（customer 情報を拡張）
versioning.register_schema("order", 2, {
    "type": "object",
    "required": ["order_id", "customer", "items"]
})

# v1 メッセージ
order_v1 = {
    "version": 1,
    "order_id": "ORD-001",
    "customer_id": "CUST-123",
    "items": [{"product_id": "PROD-A", "quantity": 2}]
}

# v2 へ移行
order_v2 = versioning.migrate(order_v1, from_version=1, to_version=2)
print(f"移行後：{order_v2}")

【アンチパターン】

✗ キューの過多（数百〜数千キュー）
   → トピックベースのルーティングを検討

✗ メッセージサイズの肥大化（1MB+）
   → 参照 ID のみを送信、実データはオブジェクトストア

✗ 同期処理の混在
   → メッセージ処理は完全に非同期に

✗ エラーハンドリングの欠如
   → DLQ + アラート通知を必須化

✗ メッセージ順序への過剰依存
   → 順序保証が必要な場合、パーティションキーを適切に設計

def analyze_bottleneck(metrics: dict) -> list:
    """
    ボトルネック分析

    Args:
        metrics: 計測メトリクス

    Returns:
        recommendations: 改善提案リスト
    """
    recommendations = []

    # キュー深さのチェック
    if metrics.get('queue_depth', 0) > 10000:
        recommendations.append(
            "キュー深さが过大 → コンシューマーの並列化を検討"
        )

    # 処理時間のチェック
    if metrics.get('avg_processing_time', 0) > 1000:  # ms
        recommendations.append(
            "処理時間が長い → バッチ処理 or 非同期化を検討"
        )

    # 再配信レートのチェック
    retry_rate = metrics.get('retry_count', 0) / max(metrics.get('total_messages', 1), 1)
    if retry_rate > 0.1:  # 10% 以上
        recommendations.append(
            "再配信率が高い → エラー原因の調査と修正を優先"
        )

    # スループットのチェック
    if metrics.get('throughput', 0) < metrics.get('target_throughput', 0) * 0.8:
        recommendations.append(
            "スループット不足 → パーティション増設 or バッチサイズ最適化"
        )

    return recommendations

# 使用例
metrics = {
    'queue_depth': 15000,
    'avg_processing_time': 1500,  # 1.5 秒
    'retry_count': 500,
    'total_messages': 10000,
    'throughput': 800,
    'target_throughput': 1000
}

issues = analyze_bottleneck(metrics)
print("ボトルネック分析結果:")
for issue in issues:
    print(f"  - {issue}")