クラス DefaultMessageListenerContainer

メソッドの詳細

• setTaskExecutor

  public void setTaskExecutor(ExecutorSE taskExecutor)

  リスナースレッドの実行に使用する Spring TaskExecutor を設定します。

  デフォルトは SimpleAsyncTaskExecutor で、指定された同時コンシューマーの数に応じて、いくつかの新しいスレッドを起動します。

  既存のスレッドプールと統合するための代替 TaskExecutor を指定します。これは、スレッドが特定の方法（Jakarta EE 環境内など）で管理されている場合にのみ、実際に価値をもたらすことに注意してください。このリスナコンテナーはその存続期間全体でいくつかのスレッドを占有するため、プレーンスレッドプールはあまり価値がありません。

  関連事項:

  • setConcurrentConsumers(int)
  • SimpleAsyncTaskExecutor

• setBackOff

  public void setBackOff(BackOff backOff)

  リカバリ試行の間隔を計算するために使用する BackOff インスタンスを指定します。BackOffExecution 実装が BackOffExecution.STOP を返す場合、このリスナーコンテナーはそれ以上の回復を試みません。

  このプロパティが設定されている場合、recovery interval は無視されます。

  導入:

  4.1

• setRecoveryInterval

  public void setRecoveryInterval(long recoveryInterval)

  回復試行の間隔をミリ秒単位で指定します。デフォルトは 5000 ミリ秒、つまり 5 秒です。これは、指定された間隔で FixedBackOff を作成する便利なメソッドです。

  その他の回復オプションについては、代わりに BackOff インスタンスを指定することを検討してください。

  関連事項:

  • setBackOff(BackOff)
  • handleListenerSetupFailure(java.lang.Throwable, boolean)

• setCacheLevelName

  public void setCacheLevelName(StringSE constantName) throws IllegalArgumentExceptionSE

  このリスナーコンテナーが適用できるキャッシュのレベルを、対応する定数の名前の形式で指定します。たとえば、"CACHE_CONNECTION"。

  例外:

  IllegalArgumentExceptionSE

  関連事項:

  • setCacheLevel(int)
  • CACHE_NONE
  • CACHE_CONNECTION
  • CACHE_SESSION
  • CACHE_CONSUMER
  • CACHE_AUTO

• setCacheLevel

  public void setCacheLevel(int cacheLevel)

  このリスナーコンテナーが適用できるキャッシュのレベルを指定します。

  外部トランザクションマネージャーが指定されている場合（外部トランザクションのスコープ内ですべてのリソースを新しく取得するため）、デフォルトは CACHE_NONE です。それ以外の場合（ローカル JMS リソースで動作する）CACHE_CONSUMER です。

  一部の Jakarta EE サーバーは、新しく取得された JMS Connection および Session の場合、進行中の XA トランザクションにのみ JMS リソースを登録します。そのため、このリスナーコンテナーはデフォルトでこれらのいずれもキャッシュしません。ただし、トランザクションリソースのキャッシュに関するサーバーのルールによっては、この設定を外部トランザクションマネージャーと組み合わせても、少なくとも CACHE_CONNECTION または CACHE_SESSION に切り替えることを検討してください。

  関連事項:

  • CACHE_NONE
  • CACHE_CONNECTION
  • CACHE_SESSION
  • CACHE_CONSUMER
  • CACHE_AUTO
  • setCacheLevelName(java.lang.String)
  • AbstractPollingMessageListenerContainer.setTransactionManager(org.springframework.transaction.PlatformTransactionManager)

• getCacheLevel

  public int getCacheLevel()

  このリスナーコンテナーが適用を許可されているキャッシュのレベルを返します。

• setConcurrency

  public void setConcurrency(StringSE concurrency)

  "lower-upper" 文字列を使用して同時実行制限を指定します。"5-10"、または単純な上限文字列。"10" （この場合、下限は 1 になります）。

  このリスナーコンテナーは常に最小数のコンシューマー（setConcurrentConsumers(int)）を保持し、負荷が増加した場合に最大数のコンシューマー setMaxConcurrentConsumers(int) に徐々に拡大します。

  次で指定:

  クラス AbstractMessageListenerContainer の setConcurrency 

• setConcurrentConsumers

  public void setConcurrentConsumers(int concurrentConsumers)

  作成する同時コンシューマーの数を指定します。デフォルトは 1 です。

  この設定に高い値を指定すると、実行時にスケジュールされる同時コンシューマーの標準レベルが増加します。これは、実質的に、任意の時点でスケジュールされる同時コンシューマーの最小数です。これは静的な設定です。動的スケーリングの場合は、代わりに "maxConcurrentConsumers" 設定を指定することを検討してください。

  同時コンシューマーの数を増やすことは、キューから入ってくるメッセージの消費を拡大するために推奨されます。ただし、複数のコンシューマーが登録されると、順序付けの保証は失われます。一般に、少量のキューの場合は 1 つのコンシューマーに固執します。

  ベンダー固有のセットアップ手段で明確に許可されていない限り、トピックの同時コンシューマーの数を上げないでください。通常のセットアップでは、これにより同じメッセージが同時に消費されることになり、これはほとんど望ましくありません。

  この設定は、JMX などを介して実行時に変更できます。

  関連事項:

  • setMaxConcurrentConsumers(int)

• getConcurrentConsumers

  public final int getConcurrentConsumers()

  "concurrentConsumer" 設定を戻します。

  これは、現在構成されている "concurrentConsumers" 値を返します。現在スケジュールされている / アクティブなコンシューマーの数は異なる場合があります。

  関連事項:

  • getScheduledConsumerCount()
  • getActiveConsumerCount()

• setMaxConcurrentConsumers

  public void setMaxConcurrentConsumers(int maxConcurrentConsumers)

  作成する同時コンシューマーの最大数を指定します。デフォルトは 1 です。

  この設定が "concurrentConsumers" より大きい場合、十分な数の受信メッセージが検出されると、リスナーコンテナーは実行時に新しいコンシューマーを動的にスケジュールします。負荷が再び低下すると、コンシューマーの数は再び標準レベル ( "concurrentConsumers" ) に削減されます。

  同時コンシューマーの数を増やすことは、キューから入ってくるメッセージの消費を拡大するために推奨されます。ただし、複数のコンシューマーが登録されると、順序付けの保証は失われます。一般に、少量のキューの場合は 1 つのコンシューマーに固執します。

  ベンダー固有のセットアップ手段で明確に許可されていない限り、トピックの同時コンシューマーの数を上げないでください。通常のセットアップでは、これにより同じメッセージが同時に消費されることになり、これはほとんど望ましくありません。

  この設定は、JMX などを介して実行時に変更できます。

  関連事項:

  • setConcurrentConsumers(int)

• getMaxConcurrentConsumers

  public final int getMaxConcurrentConsumers()

  "maxConcurrentConsumer" 設定を戻します。

  これは、現在構成されている "maxConcurrentConsumers" 値を返します。現在スケジュールされている / アクティブなコンシューマーの数は異なる場合があります。

  関連事項:

  • getScheduledConsumerCount()
  • getActiveConsumerCount()

• setMaxMessagesPerTask

  public void setMaxMessagesPerTask(int maxMessagesPerTask)

  1 つのタスクで処理するメッセージの最大数を指定します。より具体的には、これはタスクごとのメッセージ受信試行回数を制限します。これには、タイムアウトに達するまで実際にメッセージを取得しなかった受信反復が含まれます（"receiveTimeout" プロパティを参照）。

  標準の TaskExecutor の場合、デフォルトは無制限（-1）であり、（動的スケジューリングが制限される代わりに）シャットダウンするまで元の呼び出しスレッドを再利用します。

  存続期間の短いタスクの優先順位を示す SchedulingTaskExecutor の場合、デフォルトは代わりに 10 です。ここでは、長命のタスクと短命のタスクのバランスを取るために、10 〜 100 のメッセージ数を指定します。

  存続期間の長いタスクは、同じスレッドをずっと使用し続けることにより、スレッドコンテキストの頻繁な切り替えを回避します。スレッドプールは通常、短期間のタスクを優先します。

  この設定は、JMX などを介して実行時に変更できます。

  関連事項:

  • setTaskExecutor(java.util.concurrent.Executor)
  • AbstractPollingMessageListenerContainer.setReceiveTimeout(long)
  • SchedulingTaskExecutor.prefersShortLivedTasks()

• getMaxMessagesPerTask

  public final int getMaxMessagesPerTask()

  1 つのタスクで処理するメッセージの最大数を返します。

• setIdleConsumerLimit

  public void setIdleConsumerLimit(int idleConsumerLimit)

  常にアイドル状態を許可するコンシューマーの数の制限を指定します。

  この制限は、scheduleNewInvokerIfAppropriate() メソッドによって使用され、新しい呼び出し元を作成する必要があるかどうかを決定します。制限を増やすと、呼び出し元がより積極的に作成されます。これは、呼び出し元の数をより早く増やすのに役立ちます。

  デフォルトは 1 で、既存の呼び出し元がどれも現在アイドル状態でない場合にのみ、新しい呼び出し元（最初はアイドルである可能性が高い）をスケジュールするだけです。

• getIdleConsumerLimit

  public final int getIdleConsumerLimit()

  アイドル状態のコンシューマーの数の制限を返します。

• setIdleTaskExecutionLimit

  public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit)

  実行中にメッセージを受信していない、コンシューマータスクのアイドル実行の制限を指定します。この制限に達すると、タスクはシャットダウンし、受信を他の実行中のタスクに任せます。

  デフォルトは 1 で、タスクがメッセージを受信しなかった場合、アイドルリソースを早期に閉じます。これは動的スケジューリングにのみ適用されます。"maxConcurrentConsumers" 設定を参照してください。最小数のコンシューマー（"concurrentConsumers" を参照）は、いずれにしてもシャットダウンするまで保持されます。

  各タスク実行内で、メッセージ受信試行回数 ( "maxMessagesPerTask" 設定による) はそれぞれ受信メッセージ ( "receiveTimeout" 設定による) を待機します。特定のタスクでの受信試行がすべてメッセージなしで返された場合、そのタスクは受信メッセージに関してアイドル状態であるとみなされます。このようなタスクは再スケジュールできますが、指定された "idleTaskExecutionLimit" に達するとシャットダウンされます (動的スケーリングの場合)。

  スケールアップとスケールダウンが頻繁に発生する場合は、この制限を上げてください。この制限を高くすると、アイドル状態のコンシューマーが長く保持され、新しいメッセージのロードが入ってきてもコンシューマーが再起動されるのを回避できます。または、"maxMessagesPerTask" や "receiveTimeout" の値を大きくすると、アイドル状態のコンシューマーが長く保持されるようになります (スケジュールされた各タスクの平均実行時間も長くなります)。

  この設定は、JMX などを介して実行時に変更できます。

  関連事項:

  • setMaxMessagesPerTask(int)
  • AbstractPollingMessageListenerContainer.setReceiveTimeout(long)

• getIdleTaskExecutionLimit

  public final int getIdleTaskExecutionLimit()

  コンシューマータスクのアイドル実行の制限を返します。

• setIdleReceivesPerTaskLimit

  public void setIdleReceivesPerTaskLimit(int idleReceivesPerTaskLimit)

  指定されたアイドル受信数に達した後、コンシューマーを「アイドル」としてマークします。アイドル状態の受信は、潜在的な AbstractPollingMessageListenerContainer.setReceiveTimeout(long) が経過した後、受信者から null メッセージが返された瞬間からカウントされます。これにより、アイドルタスク数が setIdleTaskExecutionLimit(int) を超えているかどうかを確認し、それに基づいてタスクを再スケジュールする必要があるかどうかを判断し、それ以外の場合に保持されるリソースを節約できます。

  この設定は、受信したメッセージが null または null 以外のメッセージであるかどうかに関係なく、この制限に達した後にタスクが解放されて再スケジュールされる setMaxMessagesPerTask(int) とは異なります。この設定だけでは、タスクごとに十分な大きさのバッチが必要であるが、処理するメッセージがなくなった瞬間からクイック（er）リリースが必要な場合、柔軟性がなくなる可能性があります。

  この設定は、アイドルとしてマークされた回数の反復後にこの制限がタスクを解放することを決定する setIdleTaskExecutionLimit(int) とは異なります。

  次に例を示します: setMaxMessagesPerTask(int) が "500" に設定され、#setIdleReceivesPerTaskLimit が "60" に設定され、AbstractPollingMessageListenerContainer.setReceiveTimeout(long) が "1000" に設定され、setIdleTaskExecutionLimit(int) が "1" に設定されている場合、後続の 60 アイドル数がない限り、タスクごとに 500 メッセージが処理されます。メッセージを受信すると、タスクはアイドルとしてマークされ、解放されます。これは、最後のメッセージが処理された後、新しいメッセージが表示されない限り、60 秒後にタスクが解放されることも意味します。

  導入:

  5.3.5

  関連事項:

  • setMaxMessagesPerTask(int)
  • AbstractPollingMessageListenerContainer.setReceiveTimeout(long)

• getIdleReceivesPerTaskLimit

  public int getIdleReceivesPerTaskLimit()

  コンシューマーを「アイドル」としてマークする前に、1 つのタスクで受信する後続の null メッセージの最大数を返します。

  導入:

  5.3.5

• initialize

  public void initialize()

  クラスからコピーされた説明: AbstractJmsListeningContainer

  このコンテナーを初期化します。

  JMS 接続を作成し、ConnectionEE を開始し（"autoStartup" がオフになっていない場合）、AbstractJmsListeningContainer.doInitialize() を呼び出します。

  オーバーライド:

  クラス AbstractPollingMessageListenerContainer の initialize 

• doInitialize

  protected void doInitialize() throws JMSExceptionEE

  JMS セッションと関連する MessageConsumer の形式で、個別のスレッドで実行されている指定数の同時コンシューマーを作成します。

  次で指定:

  クラス AbstractJmsListeningContainer の doInitialize 

  例外:

  JMSExceptionEE - 登録に失敗した場合

  関連事項:

  • scheduleNewInvoker()
  • setTaskExecutor(java.util.concurrent.Executor)

• doShutdown

  protected void doShutdown() throws JMSExceptionEE

  登録済みの JMS セッションと関連する MessageConsumers を破棄します。

  次で指定:

  クラス AbstractJmsListeningContainer の doShutdown 

  例外:

  JMSExceptionEE - シャットダウンに失敗した場合

  関連事項:

  • AbstractJmsListeningContainer.shutdown()

• start

  public void start() throws JmsException

  停止コールバックがあれば、それをリセットするためにオーバーライドされます。

  次で指定:

  インターフェース Lifecycle の start 

  オーバーライド:

  クラス AbstractJmsListeningContainer の start 

  例外:

  JmsException - 起動に失敗した場合

  関連事項:

  • AbstractJmsListeningContainer.doStart()

• stop

  public void stop(RunnableSE callback) throws JmsException

  このリスナーコンテナーを停止し、すべてのリスナー処理が実際に停止したら、特定のコールバックを呼び出します。

  メモ: さらに stop(runnable) 呼び出し（処理が実際に停止する前）は、指定されたコールバックをオーバーライドします。最後に指定されたコールバックのみが呼び出されます。

  後続の start() 呼び出しが完全に停止する前にリスナーコンテナーを再起動した場合、コールバックはまったく呼び出されません。

  パラメーター:

  callback - リスナー処理が完全に停止したときに呼び出すコールバック

  例外:

  JmsException - 停止に失敗した場合

  関連事項:

  • AbstractJmsListeningContainer.stop()

• getScheduledConsumerCount

  public final int getScheduledConsumerCount()

  現在スケジュールされているコンシューマーの数を返します。

  この数値は常に "concurrentConsumers" と "maxConcurrentConsumers" の間になりますが、"activeConsumerCount" よりも高くなる場合があります (一部のコンシューマーがスケジュールされているが、現時点では実行されていない場合)。

  関連事項:

  • getConcurrentConsumers()
  • getMaxConcurrentConsumers()
  • getActiveConsumerCount()

• getActiveConsumerCount

  public final int getActiveConsumerCount()

  現在アクティブなコンシューマーの数を返します。

  この数値は常に "concurrentConsumers" と "maxConcurrentConsumers" の間になりますが、"scheduledConsumerCount" より低くなる可能性があります (一部のコンシューマーがスケジュールされているが、現時点では実行されていない場合)。

  関連事項:

  • getConcurrentConsumers()
  • getMaxConcurrentConsumers()
  • getActiveConsumerCount()

• isRegisteredWithDestination

  public boolean isRegisteredWithDestination()

  少なくとも 1 つのコンシューマーがターゲット宛先への固定登録を行ったかどうかを返します。これは、公開しようとしているメッセージを見逃さないことが保証されている実際のコンシューマーを登録することが重要である pub-sub の場合に特に興味深いものです。

  このメソッドは、start() 呼び出しの後、コンシューマーの非同期登録が発生するまでポーリングされます。このとき、メソッドは true を返し始めます。ただし、リスナーコンテナーが実際に固定登録を確立する場合に限ります。その後、コンテナーは少なくとも 1 つのコンシューマー登録を保持するため、シャットダウンするまで true を返し続けます。

  リスナーコンテナーは、最初に固定登録を持つことにバインドされていないことに注意してください。また、呼び出し元の実行ごとにコンシューマーを再作成し続ける場合もあります。これは特に cache level 設定に依存します。CACHE_CONSUMER のみが固定登録につながります。

• createDefaultTaskExecutor

  protected TaskExecutor createDefaultTaskExecutor()

  デフォルトの TaskExecutor を作成します。明示的な TaskExecutor が指定されていない場合に呼び出されます。

  デフォルトの実装では、指定された Bean 名（または Bean 名が指定されていない場合はクラス名）をスレッド名の接頭辞として使用して SimpleAsyncTaskExecutor を構築します。

  関連事項:

  • SimpleAsyncTaskExecutor(String)

• sharedConnectionEnabled

  protected final boolean sharedConnectionEnabled()

  "cacheLevel" 設定に応じて共有 JMS 接続を使用します。

  次で指定:

  クラス AbstractJmsListeningContainer の sharedConnectionEnabled 

  関連事項:

  • setCacheLevel(int)
  • CACHE_CONNECTION

• doRescheduleTask

  protected void doRescheduleTask(ObjectSE task)

  このリスナーコンテナーの TaskExecutor を介して、指定されたタスクを再実行します。

  オーバーライド:

  クラス AbstractJmsListeningContainer の doRescheduleTask 

  パラメーター:

  task - 再スケジュールするタスクオブジェクト

  関連事項:

  • setTaskExecutor(java.util.concurrent.Executor)

• messageReceived

  protected void messageReceived(ObjectSE invoker, SessionEE session)

  メッセージが入ってくることがわかっているため、新しい呼び出し側のスケジューリングを試みます...

  オーバーライド:

  クラス AbstractPollingMessageListenerContainer の messageReceived 

  パラメーター:

  invoker - 呼び出し元オブジェクト (通りました)

  session - 受信 JMS セッション

  関連事項:

  • scheduleNewInvokerIfAppropriate()

• noMessageReceived

  protected void noMessageReceived(ObjectSE invoker, SessionEE session)

  影響を受ける呼び出し元をアイドルとしてマークします。

  オーバーライド:

  クラス AbstractPollingMessageListenerContainer の noMessageReceived 

  パラメーター:

  invoker - 呼び出し元オブジェクト (通りました)

  session - 受信 JMS セッション

• scheduleNewInvokerIfAppropriate

  protected void scheduleNewInvokerIfAppropriate()

  新しい呼び出し元をスケジュールし、このリスナーコンテナーのスケジュールされた呼び出し元の合計数を増やします。ただし、指定された "maxConcurrentConsumers" 制限にまだ達しておらず、指定された "idleConsumerLimit" にも達していない場合に限ります。

  最初にメッセージを受信した呼び出し元でメッセージを処理している間にスケールアップするために、メッセージが受信されると呼び出されます。

  関連事項:

  • setTaskExecutor(java.util.concurrent.Executor)
  • getMaxConcurrentConsumers()
  • getIdleConsumerLimit()

• establishSharedConnection

  protected void establishSharedConnection()

  初期設定での失敗を受け入れるためにオーバーライドされます。最初のアクセスで共有接続を確立するのは非同期呼び出し側に任されます。

  オーバーライド:

  クラス AbstractJmsListeningContainer の establishSharedConnection 

  関連事項:

  • refreshConnectionUntilSuccessful()

• startSharedConnection

  protected void startSharedConnection()

  この実装は、Connection.start() から例外がスローされた後でも続行され、適切な回復を実行するためにリスナーに依存します。

  オーバーライド:

  クラス AbstractJmsListeningContainer の startSharedConnection 

  関連事項:

  • Connection.start()EE

• stopSharedConnection

  protected void stopSharedConnection()

  この実装は、Connection.stop() から例外がスローされた後でも続行され、再起動後に適切なリカバリを実行するためにリスナーに依存します。

  オーバーライド:

  クラス AbstractJmsListeningContainer の stopSharedConnection 

  関連事項:

  • Connection.start()EE

• handleListenerSetupFailure

  protected void handleListenerSetupFailure(ThrowableSE ex, boolean alreadyRecovered)

  リスナーのセットアップ中に発生した特定の例外を処理します。すべての同時リスナーでそのようなすべての例外が呼び出されます。

  デフォルトの実装では、まだ回復されていない場合は警告レベルで、すでに回復されている場合はデバッグレベルで例外がログに記録されます。サブクラスでオーバーライドできます。

  パラメーター:

  ex - 処理する例外

  alreadyRecovered - 以前に実行中のリスナーが、現在のリスナーのセットアップの失敗からすでに回復しているかどうか (これは通常、デバッグログ以外の目的で無視できるフォローアップエラーを示します。)

  関連事項:

  • recoverAfterListenerSetupFailure()

• recoverAfterListenerSetupFailure

  protected void recoverAfterListenerSetupFailure()

  リスナーがそれ自体のセットアップに失敗した後、基礎となる接続の再確立など、このリスナーコンテナーを回復します。

  デフォルトの実装は、DefaultMessageListenerContainer のリカバリ可能な refreshConnectionUntilSuccessful() メソッドに委譲します。このメソッドは、共有接続と非共有接続の両方のケースで JMS プロバイダーへの接続の再確立を試みます。

  関連事項:

  • refreshConnectionUntilSuccessful()
  • refreshDestination()

• refreshConnectionUntilSuccessful

  protected void refreshConnectionUntilSuccessful()

  試行が成功する前に戻るのではなく、基になる接続をリフレッシュします。共有接続の場合と共有接続なしの場合に呼び出されるため、共有接続または検証目的で確立されたばかりの一時接続で操作する必要があります。

  デフォルトの実装は、このメッセージリスナーコンテナーが実行されている限り、接続が正常に確立されるまで再試行します。再試行の間に指定された回復間隔を適用します。

  関連事項:

  • setRecoveryInterval(long)
  • start()
  • AbstractJmsListeningContainer.stop()

• refreshDestination

  protected void refreshDestination()

  このリスナーコンテナーが動作する JMS 宛先をリフレッシュします。

  キャッシュされた Destination オブジェクトが無効になったと想定して、リスナーのセットアップが失敗した後に呼び出されます（WebLogic JMS の一般的なケース）。

  デフォルトの実装では、CachingDestinationResolver の場合、DestinationResolver のキャッシュから宛先を削除します。

  関連事項:

  • AbstractMessageListenerContainer.setDestinationName(java.lang.String)
  • CachingDestinationResolver

• applyBackOffTime

  protected boolean applyBackOffTime(BackOffExecution execution)

  指定された BackOffExecution を使用して、次のバックオフ時間を適用します。

  バックオフ期間が適用され、回復を新たに試行する必要がある場合は true を返し、それ以上試行しない場合は false を返します。

  導入:

  4.1

• isRecovering

  public final boolean isRecovering()

  このリスナーコンテナーが現在リカバリを試みているかどうかを返します。

  以前に true を返すことがわかった後、isRecovering() が false に切り替えて、回復フェーズを検出するために使用できます。

  関連事項:

  • recoverAfterListenerSetupFailure()