【ITニュース解説】Brighter V10: Configurando RocketMQ

新しいBrighter V10が、メッセージングの世界で広く使われているオープンソースのプラットフォーム「Apache RocketMQ」のサポートを開始した。これは、.NETアプリケーションを開発するシステムエンジニアにとって、システム間の連携をより堅牢で高性能なものにする大きな一歩となる。

RocketMQとは、Alibaba社が開発した、分散環境でメッセージをやり取りするための高機能なシステムだ。大量のメッセージを低い遅延で、高い信頼性を持って、かつ非常に高速に処理できる特徴を持つ。まるで大規模な郵便局のように、システムAからシステムBへメッセージを確実に届け、さらにその過程でメッセージを一時的に保存したり、複数のシステムへ同時に配信したりできる。非同期通信、イベント駆動型アーキテクチャ、ストリーム処理といった、現代の複雑なマイクロサービスシステムを構築する上で不可欠な機能を提供するため、拡張性と柔軟性の高いアプリケーション開発には最適な選択肢の一つと言えるだろう。

BrighterとRocketMQを連携させるには、いくつかの準備が必要だ。まず、開発環境として.NET 8以上が必須となる。また、RocketMQのような分散システムをローカル環境で手軽に動かすために、コンテナ技術であるPodmanまたはDockerが必要だ。プロジェクトにRocketMQとBrighterの機能を取り込むには、NuGetパッケージマネージャーを通じて必要なライブラリを追加する。Brighterの従来の書き方である「Vanilla」スタイルでは、「Paramore.Brighter.MessagingGateway.RocketMQ」でRocketMQとの連携機能を、「Paramore.Brighter.ServiceActivator.Extensions.DependencyInjection」と「Paramore.Brighter.ServiceActivator.Extensions.Hosting」でBrighterのサービスとしての実行と依存性注入を有効にする。ログ記録には「Serilog.AspNetCore」が推奨される。一方、より簡潔な書き方を実現する「Fluent」スタイルでは、「Fluent.Brighter.RocketMQ」と「Paramore.Brighter.ServiceActivator.Extensions.Hosting」を使用する。

RocketMQとの連携の前に、Brighterの基本的な概念を理解しておく必要がある。Brighterでは、システム間でやり取りするデータを「Request」と呼ぶ。Requestは、具体的な指示を意味する「コマンド」と、何かが起こったという通知を意味する「イベント」の二種類に大別される。例えば、「ユーザー登録」というコマンドは特定のユーザー登録サービスへ送られるが、「注文完了」というイベントは、在庫管理や発送通知など複数のサービスが受け取る可能性がある。これらのRequestは、IRequestというインターフェースを実装したシンプルなクラスとして定義される。

Requestの準備ができたら、次に「Message Mapper」を使って、定義したRequestオブジェクトをBrighterが内部で扱うメッセージ形式に変換する。この変換はBrighter V10からは必須ではなくなったが、独自のヘッダー情報などを細かく制御したい場合には有用だ。例えば、GreetingMapperというマッパーは、GreetingというRequestをBrighterのMessageオブジェクトに変換し、メッセージID、タイムスタンプ、送り先のトピック名、メッセージの種類（イベントかコマンドか）といった情報をヘッダーに設定し、実際のデータ（ボディ）はJSON形式でシリアライズする。

メッセージがシステムに届いた後、そのメッセージを実際に処理するのが「Request Handler」だ。ハンドラーは、特定のRequestタイプを受け取り、それに応じたビジネスロジックを実行する。GreetingHandlerの例では、GreetingというRequestが届くと、その中のNameプロパティを使って「こんにちは、[名前]」というログを出力する。このように、Brighterはメッセージの送信、変換、受信、処理という一連の流れを構造化して管理するフレームワークとなる。

BrighterとRocketMQを具体的に連携させるには、まずRocketMQクラスターへの接続設定を行う。これは、RocketMQのプロキシサーバーのアドレス（例: localhost:8081）を指定し、SSLの使用有無やリクエストのタイムアウトなどの基本的な通信設定を定義する。開発環境でRocketMQをローカルで動かす場合、VPNが有効になっていると接続に問題が発生することがあるため注意が必要だ。

次に、メッセージを受信する側の設定（コンシューマ）を行う。ここでは、どの種類のメッセージ（Request）を、どのトピックから、どのコンシューマグループとして受け取るかを定義する。AddConsumersメソッドを使って、例えばGreetingというRequestタイプを受け取るためのRocketSubscriptionを登録する。この購読設定には、購読の名前、チャネルの名前、ルーティングキー（トピック名）、そしてコンシューマグループ名を指定する。コンシューマグループは、複数のアプリケーションインスタンスが同じメッセージを重複なく処理するために使われる識別子だ。メッセージがRocketMQからBrighterへどのように運ばれるかを制御する「DefaultChannelFactory」には、RocketMQ用のファクトリを指定する。

そして、メッセージを送信する側の設定（プロデューサ）を行う。AddProducesメソッドを使って、どのRequestタイプのメッセージをどのトピックへ送信するかを登録する。ここではProducerRegistryに、GreetingというRequestタイプをgreetingというトピックに送るためのRocketMqPublicationを定義する。重要な点として、RocketMQの.NETクライアントは自動的にトピックを作成しないため、メッセージを送信する前にRocketMQ側で事前にトピックを作成しておく必要がある。

よりシンプルに設定したい場合は、Brighter Fluent APIを利用できる。これは、より直感的で読みやすいコードでBrighterとRocketMQの連携設定を行うための記法だ。AddFluentBrighterメソッドの中に、RocketMQへの接続情報、メッセージ公開の設定（UsePublications）、メッセージ購読の設定（UseSubscriptions）をチェーン形式で記述していく。これにより、必要な設定を一箇所にまとめて記述しやすくなる。

ローカル開発環境でRocketMQを試す場合、Dockerとdocker-compose.ymlファイルを使うと非常に簡単に環境を構築できる。docker-compose.ymlには、主に以下のコンポーネントが定義されている。 「nameserver」は、RocketMQクラスター内の各コンポーネントの位置情報を管理する、言わば住所録のような役割を果たす。 「broker」は、実際にメッセージを保存し、プロデューサからのメッセージを受け取り、コンシューマへメッセージを配信する、郵便局の倉庫兼仕分け所の役割を担う。 「proxy」は、外部のクライアントアプリケーション（今回の.NETアプリケーションなど）がRocketMQと通信するための窓口となる。 さらに、「create-topic」というサービスが定義されており、これはRocketMQが起動した後に、先述のgreetingトピックを自動的に作成してくれる便利な機能だ。 オプションで、「dashboard」というコンポーネンスもあり、これはWebブラウザからRocketMQの状況（トピック、メッセージ、コンシューマなど）を視覚的に監視できる管理画面を提供する。これらのコンポーネントをdocker-compose upコマンド一つで起動し、http://localhost:8080にアクセスすればダッシュボードを通じて動作状況を確認できる。

現在のBrighterのRocketMQ統合にはいくつかの制限がある。例えば、メッセージを消費する際の「Pushモード」はまだサポートされておらず、現状ではメッセージを能動的に取りに行く「Pullモード」でのみ動作する。また、RocketMQの.NET SDKの既知のバグにより、メッセージが再キューイングされた際に、そのメッセージを他のコンシューマから見えなくする時間（可視性タイムアウト）を調整できないという問題がある。これらの制限は、今後のSDKのアップデートで解消される予定であり、BrighterとRocketMQの連携はさらに進化していくだろう。

このように、Brighter V10とApache RocketMQを組み合わせることで、システムエンジニアは.NETアプリケーション内で堅牢でスケーラブルな分散メッセージングを容易に実現できる。本記事で解説した内容を基に、基本的なプロデューサとコンシューマの設定、ローカル環境の構築、そしてRocketMQの特性を理解することで、現代の複雑なシステム要件に対応できる強力なアプリケーション開発を始める準備が整う。