arch-event-driven

arch-event-driven

目的

このスキルでは、Kafka、RabbitMQ、およびイベントソーシング、CQRS、パブリッシュ／サブスクライブ、デッドレターキュー、スキーマレジストリなどの関連パターンを使用して、イベント駆動型アーキテクチャを実装します。マイクロサービス環境でリアルタイムのイベント処理を行う、スケーラブルで疎結合なシステムの設計に役立ちます。

使用する場面

マイクロサービスが直接的な依存関係を持たずにイベントへ反応する必要がある場合など、非同期通信が求められるシナリオでこのスキルを使用します。大容量のデータパイプライン、リアルタイム分析、または E コマースの注文処理や IoT データストリームのように、プロデューサーとコンシューマーの分離が不可欠な場合に適用します。ポーリングで十分な単純な同期処理には使用しないでください。

主な機能

• イベントストリーミング用の Kafka トピックとパーティションを設定します。
• Kafka にイベントを保存し、状態を再構築するイベントソーシングを実装します。
• CQRS を適用して読み取りモデルと書き込みモデルを分離し、コマンドとクエリに Kafka を使用します。
• Kafka のコンシューマーグループを使用して、ファンアウトシナリオのパブリッシュ／サブスクライブを管理します。
• Kafka または RabbitMQ のデッドレターキューを使用して障害に対処します。
• Confluent Schema Registry を使用して Avro スキーマの検証を徹底します。

使用パターン

パブリッシュ／サブスクライブを実装するには、Kafka トピックを作成し、プロデューサーにイベントを発行させ、コンシューマーはグループ経由でサブスクライブします。イベントソーシングでは、すべての状態変更を Kafka ストリームのイベントとして保存し、それらを再生して現在の状態を構築します。CQRS では、コマンドを（Kafka プロデューサー経由などで）書き込みサービスに、クエリを（マテリアライズドビューからなどで）読み取りサービスにルーティングします。Kafka トピックを設定して失敗したメッセージを転送することで、デッドレターキューを使用します。一貫性を保つため、イベントスキーマは必ず JSON または Avro 形式で定義してください。

主なコマンド／API

トピック管理には Kafka CLI を使用します。次を実行します：kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3 --replication-factor 2。イベントを生成するには、次を使用します：

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders
{"orderId": 123, "status": "placed"}

イベントを消費するには、次を使用します：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --from-beginning

RabbitMQ では、CLI からエクスチェンジとキューを宣言します：rabbitmqadmin declare exchange name=events type=fanout。API エンドポイント：Kafka REST Proxy の /topics/{topic}/messages に POST してイベントを生成します（例：curl -X POST -H "Content-Type: application/vnd.kafka.json.v2+json" --data '{"records":[{"value":{"orderId":123}}]}' http://localhost:8082/topics/orders）。ヘッダーで環境変数 $KAFKA_API_KEY を使用して認証します（例：-H "Authorization: Bearer $KAFKA_API_KEY"）。設定形式：プロデューサー設定で key.serializer=org.apache.kafka.common.serialization.StringSerializer のような Kafka プロパティファイルを使用します。

統合に関する注意

Kafka クライアントライブラリ（例：Java の kafka-clients:3.0.0）を追加して、Kafka をアプリケーションに統合します。認証情報用の環境変数を設定します：RABBITMQ_URL=amqp://user:$RABBITMQ_PASSWORD@localhost を export します。スキーマレジストリを使用する場合は、プロデューサー設定で Confluent のエンドポイントを指定します：schema.registry.url=http://localhost:8081。データベースと連携する場合は、Kafka Connect を JDBC ソースに使用します。次のような JSON ファイルで設定します：{"name": "jdbc-source", "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "connection.url": "jdbc:postgresql://localhost:5432/db"}。プロデューサーが一時的なエラー時にリトライするようにしてください。

エラー処理

障害用の別のトピックを設定して、Kafka のデッドレターキューを構成します。コンシューマーコードで例外を捕捉し、"dead-letter-topic" に生成します。例：

try { consumer.poll(Duration.ofMillis(100)); } catch (Exception e) { producer.send(new ProducerRecord("dead-letter-topic", record.value())); }

RabbitMQ では、キューをデッドレターエクスチェンジにバインドします。エラーコードやタイムスタンプなどの詳細情報を必ず記録してください。Schema Registry を使用してイベントを検証し、SchemaRegistryClient API などで無効なイベントを拒否します。Kafka の JMX などのツールでラグやエラーを監視し、コンシューマーラグが 1000 メッセージを超えた場合にアラートが出るよう設定します。

具体的な使用例

例1：基本的な Kafka Pub/Sub のセットアップ
ユーザーイベント用の Pub/Sub を設定するには、まずトピックを作成します：kafka-topics.sh --bootstrap-server localhost:9092 --create --topic user-events。イベントを生成します：

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic user-events
{"userId": 1, "action": "login"}

イベントを消費します：kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic user-events --group mygroup。これにより、プロデューサーとコンシューマーが分離され、スケーラブルなイベント処理が可能になります。

例2：イベントソーシングによる CQRS の実装
e コマースアプリでは、コマンドに Kafka を使用します：producer.send(new ProducerRecord("commands-topic", "{\"command\": \"placeOrder\", \"orderId\": 123}")) で「commands-topic」に送信します。クエリには、イベントを消費してデータベースを更新することで読み取りモデルを維持します。コンシューマーコードの例：

consumer.subscribe(Arrays.asList("events-topic"));
while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord<String, String> record : records) { updateReadModel(record.value()); } }

これにより、書き込み操作と読み取りが別々に処理され、パフォーマンスが向上します。

グラフの関係

• 関連するクラスター：se-architecture
• 関連タグ：event-driven、kafka、eventsourcing、cqrs
• リンク先：se-deployment（Kafka クラスターのセットアップ用）、se-data-pipelines（イベントストリーミング統合用）