イベント駆動アーキテクチャとは？非同期処理とメッセージングパターン

基本的な概念

- イベント: システム内で発生した重要な出来事
- イベントプロデューサー: イベントを生成するコンポーネント
- イベントコンシューマー: イベントを処理するコンポーネント
- イベントブローカー: イベントの配信を管理するミドルウェア

メリット

1. 疎結合: コンポーネント間の直接的な依存関係がない
2. スケーラビリティ: 各コンポーネントを独立してスケール可能
3. フォールトトレランス: 1つのコンポーネントの障害が全体に影響しない
4. 非同期処理: 時間のかかる処理を非同期で実行可能

RabbitMQの例

RabbitMQは、オープンソースのメッセージブローカーで、AMQP（Advanced Message Queuing Protocol）を実装しています。

RabbitMQの特徴:
- 信頼性: メッセージの永続化、確認応答（acknowledgment）による配信保証
- 柔軟性: 複数のメッセージングパターンをサポート（キュー、トピック、ファンアウトなど）
- 管理UI: Webベースの管理インターフェース
- クラスタリング: 高可用性とスケーラビリティ

基本的な使い方:
1. プロデューサーがメッセージをエクスチェンジに送信
2. エクスチェンジがルーティングキーに基づいてキューにメッセージを配信
3. コンシューマーがキューからメッセージを取得して処理
4. 処理完了後に確認応答を送信

使用例:
- 非同期タスクの処理
- 通知の送信
- ログの集約
- バッチ処理のトリガー

パブリッシュ/サブスクライブパターン

パブリッシュ/サブスクライブ（Pub/Sub）パターンは、1つのメッセージを複数のコンシューマーに配信するパターンです。

パターンの特徴:
- プロデューサー（パブリッシャー）はメッセージをトピックに送信
- 複数のコンシューマー（サブスクライバー）がトピックを購読
- 1つのメッセージが複数のコンシューマーに配信される
- プロデューサーとコンシューマーが疎結合

使用例:
- 通知システム: 注文完了イベントを複数のサービスに通知
- ログ集約: ログイベントを複数の分析システムに配信
- キャッシュの無効化: データ更新イベントを複数のキャッシュに通知
- リアルタイムダッシュボード: イベントを複数のダッシュボードに配信

メリット:
- 複数のコンシューマーに効率的に配信
- プロデューサーとコンシューマーの結合を緩和
- 新しいコンシューマーを追加しやすい
- スケーラビリティが高い

Apache Kafkaの例

Apache Kafkaは、分散イベントストリーミングプラットフォームです。

Kafkaの主要概念:

1. トピック（Topic）:
- イベントのカテゴリまたはストリーム
- 複数のパーティションに分割される

2. パーティション（Partition）:
- トピックを水平に分割した単位
- 順序保証と並列処理を可能にする
- レプリケーションにより高可用性を実現

3. プロデューサー（Producer）:
- トピックにメッセージを送信
- パーティションキーを指定してメッセージをルーティング

4. コンシューマー（Consumer）:
- トピックからメッセージを読み取る
- コンシューマーグループで負荷分散
- オフセットを管理して読み取り位置を追跡

Kafkaの特徴:
- 高いスループット: 数百万メッセージ/秒を処理可能
- 水平スケーラビリティ: パーティションを追加してスケール
- 永続的なストレージ: 設定可能な保持期間でメッセージを保持
- 順序保証: パーティション内でメッセージの順序を保証
- レプリケーション: 高可用性とデータの耐久性

使用例:
- ログの集約と分析
- リアルタイムデータパイプライン
- イベントソーシング
- ストリーム処理（Kafka Streams、Kafka Connect）

実装例

実装例

CQRSの実装では、以下のようにコマンド側とクエリ側を分離します：

コマンド側（書き込み）:
- コマンドを受け取り、ビジネスロジックを実行
- ドメインモデルを使用
- イベントを生成してイベントストアに保存
- 正規化されたデータモデル
- トランザクションの整合性を重視

クエリ側（読み取り）:
- 読み取り専用のモデル（読み取りモデル）
- クエリに最適化されたデータ構造
- 非正規化されたデータモデル
- 高速なクエリパフォーマンス
- イベントから読み取りモデルを更新（プロジェクション）

実装の流れ:
1. コマンドがコマンドハンドラーに送信される
2. コマンドハンドラーがビジネスロジックを実行し、イベントを生成
3. イベントがイベントストアに保存される
4. イベントハンドラーがイベントを処理し、読み取りモデルを更新
5. クエリが読み取りモデルからデータを取得

メリット:
- 読み取りと書き込みの独立したスケーリング
- 読み取りモデルの最適化
- 複数の読み取りモデルの構築
- クエリパフォーマンスの向上

使用例:
- 高トラフィックの読み取りが多いシステム
- 複雑なクエリ要件があるシステム
- イベントソーシングと組み合わせたシステム

イベントの命名規則

イベントの命名は、イベントの意味を明確に伝えるために重要です。

命名のベストプラクティス:

1. 過去形を使用:
- イベントは既に発生した事象を表現
- 例: OrderCreated、PaymentProcessed、UserRegistered

2. ドメインの用語を使用:
- ビジネスドメインの用語を使用
- 技術的な詳細を避ける
- 例: OrderPlaced（OrderInsertedではない）

3. 明確で具体的な名前:
- 曖昧さを避ける
- イベントの内容が名前から推測できる
- 例: OrderShipped（OrderUpdatedではない）

4. 一貫性のある命名規則:
- チーム全体で統一された規則
- 例: <Entity><Action>形式（OrderCreated）

命名の例:
- ✅ OrderCreated、OrderCancelled、OrderShipped
- ✅ PaymentProcessed、PaymentFailed
- ✅ UserRegistered、UserEmailChanged
- ❌ OrderUpdate、ProcessPayment、UserChange

イベントの構造

イベントの構造は、明確で拡張可能である必要があります。

イベントの基本構造:

{
  "eventId": "uuid",
  "eventType": "OrderCreated",
  "aggregateId": "order-123",
  "aggregateType": "Order",
  "timestamp": "2024-02-22T10:00:00Z",
  "version": 1,
  "data": {
    "orderId": "order-123",
    "customerId": "customer-456",
    "totalAmount": 1000,
    "items": [...]
  },
  "metadata": {
    "userId": "user-789",
    "correlationId": "corr-abc"
  }
}

パーティションキーの使用

パーティションキーを使用することで、関連するイベントを同じパーティションにルーティングし、順序を保証できます。

パーティションキーの選択:
- 集約ID: 同じ集約に関するイベントは同じパーティションに
- エンティティID: 同じエンティティに関するイベントは同じパーティションに
- ビジネスキー: ビジネス的に関連するイベントをグループ化

例:
- 注文IDをパーティションキーとして使用
- OrderCreated、OrderItemAdded、OrderShippedは同じパーティションに
- 注文に関するイベントの順序が保証される

Kafkaでの実装:

// プロデューサーでパーティションキーを指定
producer.send({
  topic: 'orders',
  messages: [{
    key: orderId,  // パーティションキー
    value: JSON.stringify(event)
  }]
});

べき等性の実装

べき等性を実装する方法はいくつかあります：

1. イベントIDのチェック:
- 処理済みのイベントIDを記録
- 同じイベントIDが来たら処理をスキップ
- イベントストアやデータベースに処理済みIDを保存

async function handleEvent(event) {
  // イベントIDが既に処理済みかチェック
  if (await isEventProcessed(event.eventId)) {
    return; // スキップ
  }
  
  // イベントを処理
  await processEvent(event);
  
  // イベントIDを記録
  await markEventAsProcessed(event.eventId);
}

async function cancelOrder(orderId) {
  const order = await getOrder(orderId);
  
  // 既にキャンセル済みなら何もしない
  if (order.status === 'cancelled') {
    return order;
  }
  
  // キャンセル処理
  order.status = 'cancelled';
  return await saveOrder(order);
}

電子商取引システム

電子商取引システムでのイベント駆動アーキテクチャの実装例です。

システムの構成:
- 注文サービス: 注文の作成と管理
- 在庫サービス: 在庫の管理
- 決済サービス: 決済の処理
- 配送サービス: 配送の管理
- 通知サービス: 顧客への通知
- イベントブローカー: KafkaまたはRabbitMQ

イベントフロー:

1. 注文作成:
- 顧客が注文を作成
- 注文サービスがOrderCreatedイベントを発行

2. 在庫確認:
- 在庫サービスがOrderCreatedイベントを購読
- 在庫を確認し、InventoryReservedまたはInventoryInsufficientイベントを発行

3. 決済処理:
- 決済サービスがInventoryReservedイベントを購読
- 決済を処理し、PaymentProcessedまたはPaymentFailedイベントを発行

4. 注文確定:
- 注文サービスがPaymentProcessedイベントを購読
- 注文を確定し、OrderConfirmedイベントを発行

5. 配送準備:
- 配送サービスがOrderConfirmedイベントを購読
- 配送を準備し、OrderShippedイベントを発行

6. 通知:
- 通知サービスが各イベントを購読
- 顧客にメールやSMSで通知

メリット:
- サービス間の疎結合
- 各サービスが独立してスケール可能
- 新しいサービスを追加しやすい（例: 分析サービス）
- 障害の影響範囲が限定的

実装のポイント:
- イベントの順序保証（注文IDをパーティションキーに）
- べき等性の実装
- エラーハンドリングとリトライ
- デッドレターキュー（DLQ）の実装