WebSockets、STOMP、SockJS、Spring Framework 4.0によるシンプルでスケーラブルなイベントサブスクリプション

Spring Framework 4.0 M1の最初の暫定リリースは、 SockJSのサーバー側サポートを提供しました。これは、WebSocketの最も包括的で代替的な実装です。 このフォールバックオプションは、WebSocketをサポートしていないブラウザーや、プロキシがその使用を妨げる状況で必要になります。 簡単に言えば、SockJSを使用すると、今日、WebSocketアプリケーションを構築できます。これは、とりわけ、バックアップ機能に透過的に切り替えることができます。

ただし、バックアップオプションを使用しても、問題は残ります。 ソケットはかなり低レベルの抽象化であり、Webアプリケーションの大部分は今日のソケットに適合していません。 そのため、WebSocketは、HTTPでのTCPの使用方法と同様に、WebSocketでの高レベルプロトコルの使用を基本的に許可（および推奨）するサブプロトコルメカニズムを定義します。

Spring Framework 4.0 M2の2番目の中間リリースでは、WebSocket上で高レベルのメッセージングプロトコルを使用できます。 これを実証するために、サンプルアプリケーションを分析します。

株式ポートフォリオアプリ

Githubで利用できるこのテストアプリケーションを使用すると、ユーザーはポジションのポートフォリオをアップロードしたり、株式を売買したりできます。 価格見積りを送信し、位置の更新を表示します。 これはかなり単純なアプリケーションです。 ただし、ブラウザベースのメッセージングアプリケーションで発生する可能性のある多くの一般的なタスクについて説明します。


このようなアプリケーションをどのように正確に実装しますか？ HTTPおよびRESTでは、実行する必要があることを表現するためにURLおよびHTTPメソッドに依存することに慣れています。 しかし、ソケットとメッセージの束しかありません。 メッセージの対象者とその意味を知る方法


このセマンティクスを表現するには、ブラウザとサーバーが共通のメッセージ形式に同意する必要があります。 役立つ可能性のあるプロトコルがいくつかあります。 このリリースでは、シンプルさと幅広いサポートのためにSTOMPを選択しました。

シンプル/ストリーミングテキスト指向メッセージングプロトコル（STOMP）

STOMPは非常にシンプルなメッセージングプロトコルです。 HTTPでモデル化されたフレームに基づきます。 フレームは、コマンド、オプションのヘッダー、およびオプションの本体で構成されます。

たとえば、株価ポートフォリオアプリケーションは相場を送信でき、クライアントはSUBSCRIBEフレームを送信します。ここで、「宛先」見出しは、サブスクライブする内容を正確に示します。

SUBSCRIBE id:sub-1 destination:/topic/price.stock.* 

株価が利用可能になるとすぐに、サーバーは、対応する「宛先」とサブスクリプション識別子、および「コンテンツタイプ」ヘッダーと本文を含むMESSAGEフレームを送信します。

 MESSAGE subscription:sub-1 message-id:wm2si1tj-4 content-type: application/json destination:/topic/stocks.PRICE.STOCK.NASDAQ.EMC {\"ticker\":\"EMC\",\"price\":24.19} 

これをすべてブラウザーで組み合わせるには、 stomp.jsとSockJSクライアントを使用します 。

 var socket = new SockJS('/spring-websocket-portfolio/portfolio'); var client = Stomp.over(socket); var onConnect = function() { client.subscribe("/topic/price.stock.*", function(message) { //   }); }; client.connect('guest', 'guest', onConnect); 

これは重大な進歩です！ クライアント側で標準のメッセージ形式とサポートを提供しています。

これでさらに移動できます-サーバー側に。

メッセージブローカー

メッセージブローカーは、RabbitMQ、ActiveMQなどの従来のブローカー間でメッセージが転送される典型的なサーバーソリューションです。 すべてではないにしても、ほとんどがTCPを介したSTOMPをサポートし、一部はWebSocketをサポートしますが、RabbitMQは最も先進的であり、SockJSでも動作します。

アーキテクチャは次のようになります。


これは信頼性が高くスケーラブルなソリューションですが、おそらく目の前のタスクには最適ではありません。 メッセージブローカーは一般に企業内で使用されます。 ネットワークに直接公開することは、理想的なソリューションを引き付けません。

RESTアプローチから何かを学んだ場合、システム、データベースデバイス、またはドメインモデルの実装の詳細を明らかにする価値はありません。

さらに、Java開発者は、おそらくアクセス権、検証を構成し、アプリケーションロジックを追加する必要があります。 メッセージブローカーによるアプローチでは、アプリケーションサーバーはブローカーの背後にあります。これは、ほとんどのWeb開発者が慣れているものとは大きく異なります。

これが、 socket.ioのようなライブラリが人気がある理由です。 これは単純で、Webアプリケーションのニーズを対象としています。 一方、メッセージ処理の観点からメッセージブローカーの機能を無視するべきではありません。彼らはそれが本当に得意です-難しいジレンマです。 両方の長所を活用してください。

アプリケーション+メッセージブローカー

別のアプローチは、着信メッセージを処理し、Webクライアントとブローカー間の仲介として機能するアプリケーションを作成することです。 クライアントからブローカーへのメッセージは、アプリケーションを介して送信できます。返信メッセージは、アプリケーションを介してクライアントに渡されます。 これにより、アプリケーションはメッセージのタイプとヘッダーの「宛先」を判別できます。その後、メッセージを自分で処理するか、ブローカーにリダイレクトすることが決定されます。


このアプローチを選択しました。 それをより良く説明するために、いくつかのシナリオがあります。

ポジションのポートフォリオをロードする

• 顧客はポジションのポートフォリオを要求します
• アプリケーションは、サブスクリプションのデータをロードして返すことでリクエストを処理します
• メッセージブローカーはこの対話に関与していません

株価のサブスクリプション

• 顧客がサブスクリプション要求を送信します
• アプリケーションはブローカーにメッセージを送信します
• メッセージブローカーはすべての署名済みクライアントにメッセージを送信します

株価の取得

• QuoteServiceは株価情報メッセージをブローカーに送信します
• メッセージブローカーはすべての署名済みクライアントにメッセージを送信します

取引

• 顧客が取引リクエストを送信します
• アプリケーションはそれを処理せず、すべてのトランザクションはTradeServiceを通過します
• メッセージブローカーはこの対話に関与していません

更新された位置の取得

• TradeServiceは、ブローカーのキュー内の位置の更新に関するメッセージを送信します
• メッセージブローカーは位置の更新をクライアントに送信します
• 特定のユーザーへのメッセージの送信については、以下で詳しく説明します。

厳密に言えば、メッセージブローカーの使用はオプションです。 「単純な」ボックス化された代替手段を提供します。 ただし、スケーラビリティと複数のアプリケーションサーバーで展開する場合は、メッセージブローカーの使用をお勧めします。

コードスニペット

サーバーとクライアントのコードの例を見てみましょう。

これは、portfolio.jsからのポジションのポートフォリオのリクエストです。

 stompClient.subscribe("/app/positions", function(message) { self.portfolio().loadPositions(JSON.parse(message.body)); }); 

サーバー側では、 PortfolioControllerが要求を検出し、ポジションのポートフォリオを返します。これは、Webアプリケーションで頻繁に使用される要求と応答の相互作用を示しています。 Spring Securityを使用して、WebSocketとのハンドシェイクに関連する最初の1つを含むHTTP要求を以下の例のメソッドの引数として保護するため、HttpServletRequestから取得したプリンシパルSpring Securityユーザーが渡されます。

 @Controller public class PortfolioController { // ... @SubscribeEvent("/app/positions") public List<PortfolioPosition> getPortfolios(Principal principal) { String user = principal.getName(); Portfolio portfolio = this.portfolioService.findPortfolio(user); return portfolio.getPositions(); } } 

ここで、portfolio.jsが取引リクエストを送信します。

 stompClient.send("/app/trade", {}, JSON.stringify(trade)); 

サーバー側では、PortfolioControllerは実行のためにそれを渡します。

 @Controller public class PortfolioController { // ... @MessageMapping(value="/app/trade") public void executeTrade(Trade trade, Principal principal) { trade.setUsername(principal.getName()); this.tradeService.executeTrade(trade); } } 

PortfolioControllerは、ユーザーにメッセージを送信することにより、予期しない例外を処理することもできます。

 @Controller public class PortfolioController { // ... @MessageExceptionHandler @ReplyToUser(value="/queue/errors") public String handleException(Throwable exception) { return exception.getMessage(); } } 

購読しているすべての顧客にメッセージを送信するのはどうですか？ そのため、QuoteServiceはクォータを送信します。

 @Service public class QuoteService { private final MessageSendingOperations<String> messagingTemplate; @Scheduled(fixedDelay=1000) public void sendQuotes() { for (Quote quote : this.quoteGenerator.generateQuotes()) { String destination = "/topic/price.stock." + quote.getTicker(); this.messagingTemplate.convertAndSend(destination, quote); } } } 

そのため、TradeServiceは取引の完了後にポジションの更新を送信します。

 @Service public class TradeService { // ... @Scheduled(fixedDelay=1500) public void sendTradeNotifications() { for (TradeResult tr : this.tradeResults) { String queue = "/queue/position-updates"; this.messagingTemplate.convertAndSendToUser(tr.user, queue, tr.position); } } } 

@RequestMappingため...もしそうなら、PortfolioControllerには、以前にオンラインゲームを構築した開発者がこのチケットで提案したように、Spring MVCメソッド（ @RequestMapping ）が含まれている場合があります。

また、QuoteServiceやTradeServiceと同様に、Spring MVCコントローラーメソッドもメッセージを投稿できます。

Springアプリケーションでのメッセージングサポート

長い間、 Spring Integrationは、よく知られているエンタープライズ統合パターンと軽量のメッセージングのためのファーストクラスの抽象化を提供します。 このリリースの作業中に、後者がまさに私たちが信頼する必要があるものであることに気付きました。

その結果、これを発表できることを嬉しく思います。SpringIntegrationタイプのセットを新しいSpring Frameworkモジュールに移動しました。これは予想どおりspring-messagingと呼ばれます。 Message、MessageChannel、MessageHandlerなどの主要な抽象化に加えて、モジュールには、この投稿で説明する新しい機能をサポートするためのすべての注釈とクラスが含まれています。

これを念頭に置いて、Stock Portfolioアプリケーションの内部構造の図を見てみましょう。


StompWebSocketHandlerは、着信クライアントメッセージをディスパッチチャネルに配置します。 このチャンネルには3人のチャンネル登録者がいます。 1つ目はアノテーション付きメソッドを担当し、2つ目はSTOMPブローカーにメッセージを送信し、3つ目は宛先アドレスをクライアントがサブスクライブした一意の名前のキューに変換することで個々のユーザーのメッセージを処理します。

デフォルトでは、アプリケーションは知り合いに提供される「シンプルな」ブローカーで動作します。 READMEで説明されているように、プロファイルをアクティブ化および非アクティブ化することにより、「シンプルな」ブローカーとフル機能のブローカーを選択できます。


別の可能な構成変更は、 ExecutorからReactorベースのMessageChannel実装に移行することです。 Reactorプロジェクトは最近、最初の暫定リリースをリリースし、アプリケーションとメッセージブローカー間のTCP接続の管理にも使用されています。

ここでは、 新しい Spring Security構成を含む完全なアプリケーション構成を確認できます。 また、 STSの構成のサポートの改善についても興味があります。

個々のユーザーにメッセージを送信する

サインアップしたすべてのクライアントにメッセージを送信する方法を理解するのは簡単です。チャンネルにメッセージを送信するだけです。 もう少し複雑なのは、特定のクライアントにメッセージを送信する状況です。 たとえば、例外をキャッチし、エラーメッセージを送信したいとします。 または、トランザクションの完了の確認を受け取り、ユーザーを喜ばせたい場合。

従来のメッセージングアプリケーションでは、このタスクは通常、一時的なキューを作成し、応答を意味するすべてのメッセージに「reply-to」ヘッダーを設定することで実現されます。 動作しますが、かなりかさばっています。 クライアントは、該当するすべてのメッセージに必要なヘッダーを設定することを忘れてはならず、サーバーアプリケーションはそれらを監視する必要がある場合があります。 この情報は利用できない場合があります（たとえば、メッセージングの代わりにHTTP POSTが使用されている場合）。

この要件をサポートするために、接続された各クライアントに一意のキューサフィックスを送信します。 その後、サフィックスを識別子に追加して、一意のキュー名を作成できます。

 client.connect('guest', 'guest', function(frame) { var suffix = frame.headers['queue-suffix']; client.subscribe("/queue/error" + suffix, function(msg) { //   }); client.subscribe("/queue/position-updates" + suffix, function(msg) { //    }); }); 

次に、サーバー側で、 @ReplyToUserアノテーションを@MessageExceptionHandlerマークされたメソッド（または他のメッセージ処理メソッド）に追加し、戻り値をメッセージとして送信できます。

 @MessageExceptionHandler @ReplyToUser(value="/queue/errors") public String handleException(Throwable exception) { // ... } 

TradeServiceなどの他のすべてのクラスは、messagingTemplateを使用して同じことを実現できます。

 String user = "fabrice"; String queue = "/queue/position-updates"; this.messagingTemplate.convertAndSendToUser(user, queue, position); 

どちらの場合も、正しいキュー名を復元するために、内部にユーザーキューサフィックスがあります（構成済みのUserQueueSuffixResolverを使用 ）。 現時点では、単純なリゾルバの実装は1つだけです。 ただし、ユーザーまたは別のアプリケーションサーバーが接続されているかどうかに関係なく、機能をサポートするRedisベースの実装を追加するのは非常に簡単です。

結論

これが新しい機能の有益な紹介になれば幸いです。 投稿をさらに長くする代わりに、 例を検討して、作成するアプリケーションまたは作成するアプリケーションに適用することをお勧めします。 これは、フィードバックの理想的な時期です。 9月上旬にリリース候補をどのように計画しますか。