1

私は、それが可能であるかどうか、現在私が認識していない要件を持っています。メッセージに特定のプロパティが含まれている場合、JMS メッセージのエラーを一時的に無効にしたいと考えています。現在、HornetQ をメッセージ プロバイダーとして使用しています。

例を挙げてみましょう:

キューには、次の 3 つのエントリが含まれます。

{1, "foo", "A_CATEGORY"}
{2, "bar", "B_CATEGORY"}
{9, "bof", "A_CATEGORY"}

特定の時点で、アプリは HornetQ メッセージ サーバーに、B_CATEGORY に属するメッセージを現時点で配信してはならないことを伝えることができる必要があります (たとえば、B_CATEGORY オブジェクトの基礎となるデータベースが更新されるため)。そのため、id 2 のメッセージは現時点では配信されませんが、1 と 9 はカテゴリ オブジェクトの値が異なるため配信されます。

アプリケーションをまったく再起動せずに、Java コードから実行する必要があります。これはまったく可能ですか？

ご協力いただきありがとうございます！

この問題に対する別の設計アプローチについて考えてみました。最初のキューにすべての種類のカテゴリのメッセージが含まれていると仮定しましょう (カテゴリごとにキューを作成することはできません。カテゴリがたくさんある可能性があるからです)。この「通常の」キューは、通常は構成されています (たとえば、有効期限はありませんが、DLQ があります)。

リスナーがそのようなメッセージを消費し、特定のカテゴリに属する​​メッセージを処理できないことがわかった場合、それを 2 番目のキューに入れます。このキューには、再配信の遅延と有効期限が設定されています。有効期限を十分に長く設定し (もちろん、キューがオーバーフローすることはありません)、再配信時間を短すぎないように設定すると、上記の質問に対する解決策がない場合でもうまくいくはずです。

もちろん、カテゴリを処理できない間にこれらのキュー エントリがいくつ作成されるかを計算する必要があります。また、カテゴリが利用できない場合に、それに応じて再配信を調整できるようになるまでの時間。

4

2 に答える 2

1

私が知る限り、メッセージ駆動型 Bean では不可能です。

同様の機能は、標準の JMS コンシューマーで実現できます。

 MessageConsumer c = session.createConsumer(destination);
 while ( b-category-can-be-processed ) {
     Message m = c.receive();
     // process messages until b category is OK to be processed
 }

 c.close();

 // now create a different consumer with message selector ignoring "B_CATEGORY"
 MessageConsumer c1 = session.createConsumer(destination, "Category <> 'B_CATEGORY'");
 while ( b-is-locked ) {
     Message m = c1.receive();
     // process messages until b category is locked
 }

 c1.close();
 // go to start

この例では、受信したメッセージに基づいて、いつ B を再度処理するかを判断できることを前提としています。そうでない場合は、一定時間後に通常のルーチンを再開できます。この例では、単一の実行スレッドのみも示しています。

このパスをさらに詳しく調べると、Spring のDefaultMessageListenerContainer— Spring メッセージ駆動型 Bean を見ることができます。それはまさに私が説明したことを実行できますが、はるかに高度な方法で実行できます。メッセージセレクターを使用してフィードすることができ、ライブであるため、いつでも変更できます。concurrentConsumers1 より大きい値を設定すると、複数のスレッドでもメッセージを処理します。

メッセージを処理できない間に別のキューにリダイレクトするソリューションについては、余分なトラフィックが生成されることに注意してください。最後にすべてのメッセージを処理したいですか? それらをそのままにして、適切なタイミングで取得してみませんか? 難しいかもしれない再配達の遅延を先に見積もる必要はありません。

于 2011-10-09T20:47:58.823 に答える
0

フィルターを使用してコア キュー (またはサブスクリプション) を作成し、管理 API を使用してキューを停止できます。または、組み込みで作業している場合は、Server Queue オブジェクトで一時停止することができます。

これは非常にカスタムな機能であるため、組み込みで使用するか、独自のブランチで特別な調整を行うことができます。

于 2011-10-10T14:19:29.613 に答える