型パラメーター:: K - ストリームキーとストリームフィールド型。; V - ストリーム値型。

すべてのスーパーインターフェース:: Lifecycle、Phased、SmartLifecycle

public interface StreamMessageListenerContainer<K,V extends Record<K,?>> extends SmartLifecycle

メッセージリスナーコンテナーを表すフレームワークによって使用される抽象化。外部で実装することを意図したものではありません。

作成されると、StreamMessageListenerContainer は Redis ストリームにサブスクライブし、受信 messages を消費できます。StreamMessageListenerContainer は、複数のストリーム読み取りリクエストを許可し、読み取りリクエストごとに Subscription ハンドルを返します。Subscription をキャンセルすると、最終的にバックグラウンドポーリングが終了します。メッセージは key and value serializers を使用して変換され、さまざまな直列化戦略をサポートします。
StreamMessageListenerContainer は、複数のストリーム消費モードをサポートしています。

スタンドアロン
外部 StreamOperations.acknowledge(Object, String, String...) 確認応答で Consumer を使用する }
自動確認機能付きの Consumer の使用

ストリームからの読み取りには、ポーリングと、ストリームオフセットを進めるための戦略が必要です。最初の ReadOffset に応じて、StreamMessageListenerContainer は個別の戦略を適用して次の ReadOffset を取得します。
スタンドアロン

特定のメッセージ ID を使用した ReadOffset.from(String) オフセット: 指定されたオフセットから開始し、最後に表示された message Id を使用します。
ReadOffset.lastConsumed() 最後に消費された: 最新のオフセット（$）から開始し、最後に表示された message Id を使用します。
ReadOffset.latest() 最後に消費された: 最新のオフセット（$）から開始し、後続の読み取りに最新のオフセット（$）を使用します。

Consumer の使用

特定のメッセージ ID を使用した ReadOffset.from(String) オフセット: 指定されたオフセットから開始し、最後に表示された message Id を使用します。
ReadOffset.lastConsumed() 最後に消費された: コンシューマーによって最後に消費されたメッセージ（>）から開始し、コンシューマーによって最後に消費されたメッセージ（>）を後続の読み取りに使用します。
ReadOffset.latest() 最後に消費された: 最新のオフセット（$）から開始し、後続の読み取りに最新のオフセット（$）を使用します。

メモ: ReadOffset.latest() を使用すると、ポーリングが一時停止されている間にメッセージが到着する可能性があるため、メッセージがドロップされる可能性があります。messagedId をオフセットまたは ReadOffset.lastConsumed() として使用して、メッセージが失われる可能性を最小限に抑えます。

StreamMessageListenerContainer では、別の ThreadSE で長時間実行されるポーリングタスクをフォークするために ExecutorSE が必要です。このスレッドは、ストリームメッセージをポーリングして listener callback を呼び出すイベントループとして使用されます。

StreamMessageListenerContainer タスクは、ストリーム読み取り中のエラーと listener notification を構成可能な ErrorHandler に伝播します。エラーはデフォルトで Subscription を停止します。StreamMessageListenerContainer.StreamReadRequest 用に PredicateSE を構成すると、条件付きサブスクリプションをキャンセルしたり、すべてのエラーを続行したりできます。

StreamMessageListenerContainer の使用方法については、次のサンプルコードを参照してください。

 RedisConnectionFactory factory = …;

 StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer.create(factory);
 Subscription subscription = container.receive(StreamOffset.fromStart("my-stream"), message -> …);

 container.start();

 // later
 container.stop();

導入:

2.2

作成者:

Mark Paluch, Christoph Strobl, Christian Rest

関連事項:

ネストされたクラスのサマリー
ネストされたクラス
修飾子と型
インターフェース
説明
static class
StreamMessageListenerContainer.ConsumerStreamReadRequest<K>
Consumer を使用して Redis ストリームを読み取るようにリクエストします。
static class
StreamMessageListenerContainer.ConsumerStreamReadRequestBuilder<K>
StreamMessageListenerContainer.ConsumerStreamReadRequest を構築するビルダー。
static class
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<K,V extends Record<K,?>>
StreamMessageListenerContainer のオプション。
static class
StreamMessageListenerContainer.StreamMessageListenerContainerOptionsBuilder<K,V extends Record<K,?>>
StreamMessageListenerContainer.StreamMessageListenerContainerOptions のビルダー。
static class
StreamMessageListenerContainer.StreamReadRequest<K>
Redis ストリームの読み取りをリクエストします。
static class
StreamMessageListenerContainer.StreamReadRequestBuilder<K>
StreamMessageListenerContainer.StreamReadRequest を構築するビルダー。
フィールドサマリー
インターフェース org.springframework.context.SmartLifecycle から継承されたフィールド
DEFAULT_PHASE
メソッドのサマリー
修飾子と型
メソッド
説明
static StreamMessageListenerContainer<StringSE,MapRecord<StringSE,StringSE,StringSE>>
create(RedisConnectionFactory connectionFactory)
RedisConnectionFactory を指定して string serializers を使用して新しい StreamMessageListenerContainer を作成します。
static <K, V extends Record<K, ?>> StreamMessageListenerContainer<K,V>
create(RedisConnectionFactory connectionFactory, StreamMessageListenerContainer.StreamMessageListenerContainerOptions<K,V> options)
RedisConnectionFactory および StreamMessageListenerContainer.StreamMessageListenerContainerOptions を指定して、新しい StreamMessageListenerContainer を作成します。
default Subscription
receive(Consumer consumer, StreamOffset<K> streamOffset, StreamListener<K,V> listener)
Redis ストリームの新しいサブスクリプションを登録します。
default Subscription
receive(StreamOffset<K> streamOffset, StreamListener<K,V> listener)
Redis ストリームの新しいサブスクリプションを登録します。
default Subscription
receiveAutoAck(Consumer consumer, StreamOffset<K> streamOffset, StreamListener<K,V> listener)
Redis ストリームの新しいサブスクリプションを登録します。
Subscription
register(StreamMessageListenerContainer.StreamReadRequest<K> streamRequest, StreamListener<K,V> listener)
Redis ストリームの新しいサブスクリプションを登録します。
void
remove(Subscription subscription)
特定の Subscription をコンテナーから登録解除します。
インターフェース org.springframework.context.Lifecycle から継承されたメソッド
isRunning, start, stop
インターフェース org.springframework.context.SmartLifecycle から継承されたメソッド
getPhase, isAutoStartup, stop

メソッドの詳細
- create
  static StreamMessageListenerContainer<StringSE,MapRecord<StringSE,StringSE,StringSE>> create(RedisConnectionFactory connectionFactory)
  RedisConnectionFactory を指定して string serializers を使用して新しい StreamMessageListenerContainer を作成します。
  パラメーター:
  connectionFactory - null であってはなりません。
  戻り値:
  新しい StreamMessageListenerContainer
- create
  static <K, V extends Record<K, ?>> StreamMessageListenerContainer<K,V> create(RedisConnectionFactory connectionFactory, StreamMessageListenerContainer.StreamMessageListenerContainerOptions<K,V> options)
  RedisConnectionFactory および StreamMessageListenerContainer.StreamMessageListenerContainerOptions を指定して、新しい StreamMessageListenerContainer を作成します。
  パラメーター:
  connectionFactory - null であってはなりません。
  options - null であってはなりません。
  戻り値:
  新しい StreamMessageListenerContainer
- receive
  default Subscription receive(StreamOffset<K> streamOffset, StreamListener<K,V> listener)
  Redis ストリームの新しいサブスクリプションを登録します。is already running の場合、Subscription が追加されてすぐに実行されます。それ以外の場合は、コンテナーが実際に started になると、スケジュールされて開始されます。
  Record 取得中のエラーは、基礎となるタスクの cancellation につながります。
  Lifecycle.stop() では、コンテナー自体をシャットダウンする前に、すべての subscriptions がキャンセルされます。
  パラメーター:
  streamOffset - そのオフセットに沿ったストリーム。
  listener - null であってはなりません。
  戻り値:
  サブスクリプションハンドル。
  関連事項:
  StreamOffset.create(Object, ReadOffset)
- receive
  default Subscription receive(Consumer consumer, StreamOffset<K> streamOffset, StreamListener<K,V> listener)
  Redis ストリームの新しいサブスクリプションを登録します。is already running の場合、Subscription が追加されてすぐに実行されます。それ以外の場合は、コンテナーが実際に started になると、スケジュールされて開始されます。
  すべてのメッセージは、処理後に StreamOperations.acknowledge(Object, String, String...) を使用して確認応答する必要があります。
  Record 取得中のエラーは、基礎となるタスクの cancellation につながります。
  Lifecycle.stop() では、コンテナー自体をシャットダウンする前に、すべての subscriptions がキャンセルされます。
  パラメーター:
  consumer - コンシューマーグループは、null であってはなりません。
  streamOffset - そのオフセットに沿ったストリーム。
  listener - null であってはなりません。
  戻り値:
  サブスクリプションハンドル。
  関連事項:
  StreamOffset.create(Object, ReadOffset)
  ReadOffset.lastConsumed()
- receiveAutoAck
  default Subscription receiveAutoAck(Consumer consumer, StreamOffset<K> streamOffset, StreamListener<K,V> listener)
  Redis ストリームの新しいサブスクリプションを登録します。is already running の場合、Subscription が追加されてすぐに実行されます。それ以外の場合は、コンテナーが実際に started になると、スケジュールされて開始されます。
  すべてのメッセージは、受信時に確認されます。
  Record 取得中のエラーは、基礎となるタスクの cancellation につながります。
  Lifecycle.stop() では、コンテナー自体をシャットダウンする前に、すべての subscriptions がキャンセルされます。
  パラメーター:
  consumer - コンシューマーグループは、null であってはなりません。
  streamOffset - そのオフセットに沿ったストリーム。
  listener - null であってはなりません。
  戻り値:
  サブスクリプションハンドル。
  関連事項:
  StreamOffset.create(Object, ReadOffset)
  ReadOffset.lastConsumed()
- register
  Subscription register(StreamMessageListenerContainer.StreamReadRequest<K> streamRequest, StreamListener<K,V> listener)
  Redis ストリームの新しいサブスクリプションを登録します。is already running の場合、Subscription が追加されてすぐに実行されます。それ以外の場合は、コンテナーが実際に started になると、スケジュールされて開始されます。
  Record 中のエラーは、基になるタスクをキャンセルするかどうかのテスト cancellation predicate に対してテストされます。
  Lifecycle.stop() では、コンテナー自体をシャットダウンする前に、すべての subscriptions がキャンセルされます。
  Record 取得中のエラーは、指定された StreamMessageListenerContainer.StreamReadRequest.getErrorHandler() に委譲されます。
  パラメーター:
  streamRequest - null であってはなりません。
  listener - null であってはなりません。
  戻り値:
  サブスクリプションハンドル。
  関連事項:
  StreamMessageListenerContainer.StreamReadRequest
  StreamMessageListenerContainer.ConsumerStreamReadRequest
- remove
  void remove(Subscription subscription)
  特定の Subscription をコンテナーから登録解除します。これにより、潜在的な stop/start シナリオで Subscription が再起動されるのを防ぎます。
  active subcription は、削除前は cancelled です。
  パラメーター:
  subscription - null であってはなりません。

インターフェース StreamMessageListenerContainer<K,V extends Record<K,?>>

ネストされたクラスのサマリー

フィールドサマリー

インターフェース org.springframework.context.SmartLifecycle から継承されたフィールド

メソッドのサマリー

インターフェース org.springframework.context.Lifecycle から継承されたメソッド

インターフェース org.springframework.context.SmartLifecycle から継承されたメソッド

メソッドの詳細

create

create

receive

receive

receiveAutoAck

register

remove