バッチの消費

バージョン 3.0 以降、spring.cloud.stream.bindings.<name>.consumer.batch-mode が true に設定されている場合、Kafka Consumer のポーリングによって受信されたすべてのレコードは、List<?> としてリスナーメソッドに提示されます。それ以外の場合、メソッドは一度に 1 つのレコードで呼び出されます。バッチのサイズは、Kafka コンシューマープロパティ max.poll.records、fetch.min.bytes、fetch.max.wait.ms によって制御されます。詳細については、Kafka のドキュメントを参照してください。

バッチを受信するときは、次の型署名が許可されます。

List<Person> の最初のオプションでは、リスナーはメッセージヘッダーを取得しません。2 番目の型 シグネチャー (Message<List<Person>>) が使用される場合、ヘッダーにアクセスできます。ただし、ヘッダーはすべて Collection の形式のままです。次の例を見てみましょう。

Message に 10 個の Person オブジェクトのリストが含まれているとします。Message の MessageHeaders には、キーがヘッダー名、値がリストであるヘッダーのマップが含まれています。このリストには、ペイロードリストと同じ順序でそのヘッダーのヘッダー値が含まれています。ペイロードリストの反復に基づいて MessageHeaders マップからヘッダーに正しくアクセスするのは、アプリケーションの責任です。

バッチモードで使用する場合、List<Message<Person>> 形式の型署名は許可されないことに注意してください。

バージョン 4.0.2 以降、バインダーは、バッチモードでの消費時に DLQ 機能をサポートします。バッチモードのコンシューマーバインディングで DLQ を使用する場合、前のポーリングから受信したすべてのレコードが DLQ トピックに配信されることに注意してください。