using-message-queues

メッセージキュー

イベント駆動型アーキテクチャ、バックグラウンドジョブ処理、サービスの疎結合化のための非同期通信パターンを実装します。

このスキルを使う場面

次のような場合は、メッセージキューを使用します。

• 長時間実行される処理が HTTP リクエストをブロックする場合（レポート生成、動画処理）
• サービスの疎結合化が必要な場合（マイクロサービス、イベント駆動型アーキテクチャ）
• 確実な配信が必要な場合（決済処理、注文の履行）
• 分析のためのイベントストリーミングが必要な場合（ログ集約、メトリクスパイプライン）
• 複雑なプロセスのワークフローオーケストレーションが必要な場合（複数ステップの Saga、人間参加型ワークフロー）
• バックグラウンドジョブ処理が必要な場合（メール送信、画像のリサイズ）

ブローカー選択の決定ツリー

主なニーズに応じてメッセージブローカーを選択します。

イベントストリーミング／ログ集約

→ Apache Kafka

• スループット：500K-1M msg/s
• イベントの再生（イベントソーシング）
• Exactly-once セマンティクス
• 長期保持
• 用途：分析パイプライン、CQRS、イベントソーシング

シンプルなバックグラウンドジョブ

→ タスクキュー

• Python → Celery + Redis
• TypeScript → BullMQ + Redis
• Go → Asynq + Redis
• 用途：メール送信、レポート生成、Webhook

複雑なワークフロー／Saga

→ Temporal

• 永続的な実行（再起動後も継続）
• Saga パターンのサポート
• 人間参加型ワークフロー
• 用途：注文処理、AI エージェントのオーケストレーション

リクエスト／レスポンス／RPC パターン

→ NATS

• リクエスト／レスポンス機能を標準搭載
• サブミリ秒のレイテンシ
• クラウドネイティブでシンプルな運用
• 用途：マイクロサービスの RPC、IoT のコマンド／制御

複雑なメッセージルーティング

→ RabbitMQ

• エクスチェンジ（direct、topic、fanout、headers）
• デッドレターエクスチェンジ
• メッセージ TTL、優先度
• 用途：複数コンシューマーパターン、pub/sub

すでに Redis を使用している場合

→ Redis Streams

• 新たなインフラストラクチャは不要
• シンプルなコンシューマーグループ
• 中程度のスループット（100K+ msg/s）
• 用途：通知キュー、シンプルなジョブキュー

パフォーマンス比較

ブローカー	スループット	レイテンシ（p99）	最適な用途
Kafka	500K-1M msg/s	10-50ms	イベントストリーミング
NATS JetStream	200K-400K msg/s	サブミリ秒～5ms	クラウドネイティブなマイクロサービス
RabbitMQ	50K-100K msg/s	5-20ms	タスクキュー、複雑なルーティング
Redis Streams	100K+ msg/s	サブミリ秒	シンプルなキュー、キャッシュ

クイックスタートの例

Kafka プロデューサー／コンシューマー（Python）

動作するコードについては、examples/kafka-python/ を参照してください。

from confluent_kafka import Producer, Consumer

# Producer
producer = Producer({'bootstrap.servers': 'localhost:9092'})
producer.produce('orders', key='order_123', value='{"status": "created"}')
producer.flush()

# Consumer
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processors',
    'auto.offset.reset': 'earliest'
})
consumer.subscribe(['orders'])

while True:
    msg = consumer.poll(1.0)
    if msg is not None:
        process_order(msg.value())

Celery バックグラウンドジョブ（Python）

完全な実装については、examples/celery-image-processing/ を参照してください。

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379')

@app.task(bind=True, max_retries=3)
def process_image(self, image_url: str):
    try:
        result = expensive_image_processing(image_url)
        return result
    except RecoverableError as e:
        raise self.retry(exc=e, countdown=60)

BullMQ ジョブ処理（TypeScript）

完全な実装については、examples/bullmq-webhook-processor/ を参照してください。

import { Queue, Worker } from 'bullmq'

const queue = new Queue('webhooks', {
  connection: { host: 'localhost', port: 6379 }
})

// Enqueue job
await queue.add('send-webhook', {
  url: 'https://example.com/webhook',
  payload: { event: 'order.created' }
})

// Process jobs
const worker = new Worker('webhooks', async job => {
  await fetch(job.data.url, {
    method: 'POST',
    body: JSON.stringify(job.data.payload)
  })
}, { connection: { host: 'localhost', port: 6379 } })

Temporal ワークフローオーケストレーション

Saga パターンの実装については、examples/temporal-order-saga/ を参照してください。

from temporalio import workflow, activity
from datetime import timedelta

@workflow.defn
class OrderSagaWorkflow:
    @workflow.run
    async def run(self, order_id: str) -> str:
        # Step 1: Reserve inventory
        inventory_id = await workflow.execute_activity(
            reserve_inventory,
            order_id,
            start_to_close_timeout=timedelta(seconds=10),
        )

        # Step 2: Charge payment
        payment_id = await workflow.execute_activity(
            charge_payment,
            order_id,
            start_to_close_timeout=timedelta(seconds=30),
        )

        return f"Order {order_id} completed"

コアパターン

イベントの命名規則

次の形式を使用します：Domain.Entity.Action.Version

例：

• order.created.v1
• user.profile.updated.v2
• payment.failed.v1

イベントスキーマの構造

{
  "event_type": "order.created.v2",
  "event_id": "uuid-here",
  "timestamp": "2025-12-02T10:00:00Z",
  "version": "2.0",
  "data": {
    "order_id": "ord_123",
    "customer_id": "cus_456"
  },
  "metadata": {
    "producer": "order-service",
    "trace_id": "abc123",
    "correlation_id": "xyz789"
  }
}

デッドレターキューパターン

最大再試行回数に達した後、失敗したメッセージをデッドレターキュー（DLQ）にルーティングします。

@app.task(bind=True, max_retries=3)
def process_order(self, order_id: str):
    try:
        result = perform_processing(order_id)
        return result
    except UnrecoverableError as e:
        send_to_dlq(order_id, str(e))
        raise Reject(e, requeue=False)

Exactly-once 処理のための冪等性

@app.post("/process")
async def process_payment(
    payment_data: dict,
    idempotency_key: str = Header(None)
):
    # Check if already processed
    cached_result = redis_client.get(f"idempotency:{idempotency_key}")
    if cached_result:
        return {"status": "already_processed"}

    result = process_payment_logic(payment_data)
    redis_client.setex(f"idempotency:{idempotency_key}", 86400, result)
    return {"status": "processed", "result": result}

フロントエンドとの連携

SSE によるジョブステータスの更新

# FastAPI endpoint for real-time job status
@app.get("/status/{task_id}")
async def task_status_stream(task_id: str):
    async def event_generator():
        while True:
            task = celery_app.AsyncResult(task_id)

            if task.state == 'PROGRESS':
                yield {"event": "progress", "data": task.info.get('progress', 0)}
            elif task.state == 'SUCCESS':
                yield {"event": "complete", "data": task.result}
                break

            await asyncio.sleep(0.5)

    return EventSourceResponse(event_generator())

React コンポーネント

export function JobStatus({ jobId }: { jobId: string }) {
  const [progress, setProgress] = useState(0)

  useEffect(() => {
    const eventSource = new EventSource(`/api/status/${jobId}`)

    eventSource.addEventListener('progress', (e) => {
      setProgress(JSON.parse(e.data))
    })

    eventSource.addEventListener('complete', (e) => {
      toast({ title: 'Job complete', description: JSON.parse(e.data) })
      eventSource.close()
    })

    return () => eventSource.close()
  }, [jobId])

  return <ProgressBar value={progress} />
}

詳細ガイド

包括的なドキュメントについては、リファレンスファイルを参照してください。

ブローカー別ガイド

• Kafka：パーティショニング、コンシューマーグループ、Exactly-once セマンティクスについては、references/kafka.md を参照してください
• RabbitMQ：エクスチェンジ、バインディング、ルーティングパターンについては、references/rabbitmq.md を参照してください
• NATS：JetStream、リクエスト／レスポンスパターンについては、references/nats.md を参照してください
• Redis Streams：コンシューマーグループ、確認応答については、references/redis-streams.md を参照してください

タスクキューガイド

• Celery：定期タスク、canvas（ワークフロー）、モニタリングについては、references/celery.md を参照してください
• BullMQ：ジョブの優先順位付け、フロー、Bull Board によるモニタリングについては、references/bullmq.md を参照してください
• Temporal：Saga パターン、シグナル、クエリについては、references/temporal-workflows.md を参照してください

パターンガイド

• イベントパターン：イベントソーシング、CQRS、アウトボックスパターンについては、references/event-patterns.md を参照してください。

避けるべきよくあるアンチパターン

1. 長時間の処理に同期 API を使用する

# ❌ BAD: Blocks request thread
@app.post("/generate-report")
def generate_report(user_id: str):
    report = expensive_computation(user_id)  # 5 minutes!
    return report

# ✅ GOOD: Enqueue background job
@app.post("/generate-report")
async def generate_report(user_id: str):
    task = generate_report_task.delay(user_id)
    return {"task_id": task.id}

2. 冪等性のないコンシューマー

# ❌ BAD: Processes duplicates
@app.task
def send_email(email: str):
    send_email_service(email)  # Sends twice if retried!

# ✅ GOOD: Idempotent with deduplication
@app.task
def send_email(email: str, idempotency_key: str):
    if redis.exists(f"sent:{idempotency_key}"):
        return "already_sent"
    send_email_service(email)
    redis.setex(f"sent:{idempotency_key}", 86400, "1")

3. デッドレターキューを無視する

# ❌ BAD: Failed messages lost forever
@app.task(max_retries=3)
def risky_task(data):
    process(data)  # If all retries fail, data disappears

# ✅ GOOD: DLQ for manual inspection
@app.task(max_retries=3)
def risky_task(data):
    try:
        process(data)
    except Exception as e:
        if self.request.retries >= 3:
            send_to_dlq(data, str(e))
        raise

4. リクエスト／レスポンスに Kafka を使用する

# ❌ BAD: Kafka is not designed for RPC
def get_user_profile(user_id: str):
    kafka_producer.send("user_requests", {"user_id": user_id})
    # How to correlate response? Kafka is asynchronous!

# ✅ GOOD: Use NATS request-reply or HTTP/gRPC
response = await nats.request("user.profile", user_id.encode())

ライブラリの推奨

Context7 による調査

Confluent Kafka (Python)

• Context7 ID: /confluentinc/confluent-kafka-python
• 信頼スコア：68.8/100
• コードスニペット：192+
• 本番環境に対応した Python Kafka クライアント

Temporal

• Context7 ID: /websites/temporal_io
• 信頼スコア：80.9/100
• コードスニペット：3,769+
• 耐久性のある実行を実現するワークフローオーケストレーション

インストール

Python：

pip install confluent-kafka celery[redis] temporalio aio-pika redis

TypeScript/Node.js：

npm install kafkajs bullmq @temporalio/client amqplib ioredis

Rust：

cargo add rdkafka lapin async-nats redis

Go：

go get github.com/confluentinc/confluent-kafka-go
go get github.com/hibiken/asynq
go get go.temporal.io/sdk

ユーティリティ

セットアップを自動化するには、スクリプトを使用してください。

• Kafka のセットアップ：テスト用ユーティリティを実行するには、python scripts/kafka_producer_consumer.py を実行します
• スキーマの検証：イベントスキーマを検証するには、python scripts/validate_message_schema.py を実行します

関連スキル

• api-patterns：非同期ジョブ送信のための API 設計
• realtime-sync：ジョブのステータス更新に WebSocket/SSE を使用
• feedback：ジョブ完了時のトースト通知
• databases-*：イベントログの永続ストレージ
• observability：キュー操作のトレーシングとメトリクス