異なるソースシステムと単一のコンシューマに対するアクティブなMQ同時実行

Question

私のキューには異なるソースシステムからのメッセージがあるというこの問題があります。

たとえば、最初のメッセージには「X」というソースシステム名があり、2番目のソースシステム名は「Y」です。

現在のところ、並行処理レベルが1のJMSリスナーがあります。すべてのメッセージが1つずつ処理されるようになっていますが、メッセージは同時に処理する必要があります。そのソースシステムに対して一度にメッセージを処理する必要があり、異なるソースシステム用のメッセージがある場合は、それらを並行して実行する必要があります。

ソースシステムが動的に作成されているため、ソースシステムごとにキューとコンシューマーを分離することができません。

誰かが私を正しい方向に押し込んでくれれば素晴らしいだろう。

Answer 1

あなたの問題は、特定のソースから発信されたメッセージの順序付き配信を維持することですが、さまざまなソースからのメッセージを並行して処理することができるように思えます。

message groupsを使用してこれを行うことができます。

メッセージ・アプリケーションは、ブローカーを使用して、関連する一連のメッセージをグループに属するものとして分類することができます。これにより、メッセージ・プロデューサは、メッセージ・グループがアプリケーションに対して単一の論理操作とみなされるべきであることを消費者に示すことができる。 。

この仕事をするために、生産システム生産システム名にJMSXGroupIDヘッダを設定している：

Mesasge message = session.createTextMessage("<message />"); 
message.setStringProperty("JMSXGroupID", "SourceSystem1Name"); 

その後ブローカは、そのグループに属するメッセージの中で、消費順序付けを実施します。

補遺

それらが動的に を作成しているように、ソース・システムのN番号があるかもしれませんし、キュー

に これらのNソース・システムからのすべてのメッセージを置くプロデューサーがあります

したがって、メッセージプロデューサはJMSXGroupIDヘッダーをソースシステムの名前に設定できます。

私が直面してる問題は、一つのソースシステムのためのメッセージが が処理取得されている場合は、他のソースシステムメッセージが

だから、一度そのメッセージの処理の 完了まで待つ必要がありますグループヘッダーが記述されているように設定されている場合、ブローカーは指定されたソースシステムのメッセージをコンシューマーに順番にリリースするだけです。これは、グループ内の次のメッセージを解放する前に、前のメッセージが処理されたことを確認するように消費者に強制することによって行われます。

並行性を適切な値に設定することで、コンシューマはさまざまなソースシステムからのメッセージを並行して処理できますが、必要な動作である任意のソースシステムからのメッセージを順次処理することが強制されます。

Answer 2

コンシューマはコンシューマ側で宣言されているため、別の変数に従って動的に変更することはできません。私はお勧めし

ソリューションは、セレクタを使用して、次のような2つの消費者の接続を使用することです：並列処理のための

消費者のエンドポイント（この場合は5）意志のようになります。

activemq:queue:MY_QUEUE_UNIQUE?concurrentConsumers=5&selector=MySourceHeader<>'X' 

消費者のエンドポイント連番の処理のためにルックスは次のようになります：

activemq:queue:MY_QUEUE_UNIQUE?concurrentConsumers=1&selector=MySourceHeader='X' 

これは、キューの1のフェッチサイズを定義するだけでなく良いでしょう（でも、私はSではありませんよそれは必須です）。