プロデューサー実装ロジック - Kafka 知識システム (II)

[[409180]]

Kafka はメッセージを個別に送信しますか、それともバッチで送信しますか?

Kafka はどのようにして単一のメッセージを送信するのでしょうか?

Kafka はメッセージを順番に送信しますか?

どのような状況でプロデューサーが FullGC を頻繁に実行する可能性がありますか?

メッセージ送信ロジック

メッセージ送信プロセスの鳥瞰図。

プロデューサーデザイン

消費と送信のメカニズム:

1) シリアライザー: メッセージオブジェクトをバイト配列にシリアル化し、ネットワーク経由で送信します。

2) パーティショナー: メッセージが送信される特定のパーティションを計算します。パーティションが明示的に指定されている場合、パーティショナーは使用されません。

3) メッセージバッファプール: クライアントのメッセージバッファプール。デフォルトサイズは 32M です。パラメータ buffer.memory を参照してください。

4) バッチ送信: バッファプール内のメッセージはバッチで送信されます。デフォルトのバッチサイズは 16 KB です。パラメータ batch.size を参照してください。

負荷分散設計:

メッセージトピックは複数のパーティションで構成され、パーティションは異なるブローカーに均等に分散されます。したがって、ブローカークラスターのパフォーマンスを効果的に活用し、メッセージのスループットを向上させるために、プロデューサーはランダム方式またはハッシュ方式を使用してメッセージを複数のパーティションに均等に送信し、負荷分散を実現できます。

パーティション分割戦略:

ポーリング戦略、デフォルト戦略
ランダム戦略は、実際のパフォーマンスでは、ポーリング戦略よりも劣っています。
メッセージキー順序保持戦略によれば、メッセージがキーで定義されると、各パーティションでのメッセージ処理は順次行われるため、同じキーを持つすべてのメッセージが同じパーティションに入ることが保証されます。

カフカプロデューサー

ソースコード

//クライアントID。 KafkaProducer を作成するときに、client.id を通じて clientId を定義できます。指定しない場合は、デフォルトはproducer-seqになります。 seq はプロセス中に増加します。クライアントが clientId を明示的に指定することを強くお勧めします。
 プライベート最終文字列クライアントID;
   // メッセージ サイズ、送信時間、その他の監視関連のメトリックなどのメトリックのストレージ コンテナー。
    最終的なメトリクス メトリクス。
    //パーティション負荷分散アルゴリズム。パラメータpartitioner.classで指定されます。
    プライベート最終パーティショナー パーティショナー。
    //キーとメッセージ本文のシリアル化後のメッセージの合計サイズを含む、send メソッドの呼び出しによって送信される最大リクエスト サイズは、この値を超えることはできません。パラメータmax.request.sizeで設定します。
    プライベート最終int maxRequestSize;
    //プロデューサー キャッシュが占有するメモリの合計サイズは、パラメーター buffer.memory によって設定されます。
    プライベート最終長い合計メモリサイズ;
    //トピックルーティング情報などのメタデータ情報は、KafkaProducer によって自動的に更新されます。
    プライベート最終メタデータメタデータ;
    //メッセージレコードアキュムレータ
    プライベート最終 RecordAccumulator アキュムレータ;
    // メッセージを送信するロジック、つまりブローカーにメッセージを送信する処理ロジックをカプセル化するために使用されます。
    プライベート最終送信者送信者;
    // メッセージを送信するためのバックグラウンド スレッド。ブローカーにメッセージを送信するために内部的に Sender を使用する独立したスレッド。
    プライベート最終スレッド ioThread;
    //圧縮タイプ。圧縮はデフォルトでは有効になっていませんが、パラメータ compression.type を介して設定できます。オプションの値: none、gzip、snappy、lz4、zstd。
    プライベート最終圧縮タイプ圧縮タイプ;
    // 監視のメトリックとして使用されるエラー情報コレクター。
    プライベート最終センサーエラー;
    // システム時間やスレッドのスリープなどを取得するために使用されます。
    プライベート最終時間 時間;
    //メッセージのキーをシリアル化するために使用されます。
    プライベート最終ExtendedSerializer<K> keySerializer;
    //シリアライザー< V> 値シリアライザー
    プライベート最終ExtendedSerializer<V> valueSerializer;
    //プロデューサーの構成情報。
    プライベート最終プロデューサーConfigプロデューサーConfig;
    //最大ブロック時間。プロデューサーが使用するキャッシュが指定された値に達すると、メッセージの送信がブロックされます。最大待機時間は、パラメータmax.block.msで設定できます。
    プライベート最終long maxBlockTimeMs;
   // クライアントがリクエストへの応答を待機する最大時間を構成します。タイムアウトが経過する前に応答が受信されない場合、クライアントは必要に応じて要求を再送信するか、再試行回数が上限に達した場合は要求を失敗させます。
    プライベート最終int requestTimeoutMs;
    // プロデューサー側のインターセプターは、メッセージが送信される前にいくつかのカスタマイズされた処理を実行します。
    プライベート最終 ProducerInterceptors<K, V> インターセプター;
    //API バージョンの関連メタ情報を維持します。このクラスは Kafka 内でのみ使用できます。
    プライベート最終 ApiVersions apiVersions;
    //Kafka メッセージ トランザクション マネージャー。
    プライベート最終トランザクションマネージャトランザクションマネージャ;
    //Kafka プロデューサー トランザクション コンテキスト環境の初期結果。
    プライベート TransactionalRequestResult initTransactionsResult;

KafkaProducer には次の特性があります。

KafkaProducer はスレッドセーフであり、複数のスレッドで使用できます。
KafkaProducer には、送信するメッセージを格納するためのキャッシュプール、つまり ProducerRecord キューが含まれています。同時に、ProducerRecord オブジェクトを Kafka クラスターに送信するための IO スレッドが開始されます。
KafkaProducer のメッセージ送信 API の send メソッドは非同期です。送信するメッセージの ProducerRecord をバッファ領域に送信し、すぐに戻って結果証明書 Future を返すことだけを担当します。

acksパラメータの役割

KafkaProducer は、メッセージが「送信」される条件 (標準) を定義するコアパラメータ acks を提供します。これは、ブローカー側がクライアントに対してメッセージが送信されたことを約束する条件です。オプションの値は次のとおりです。

0: KafkaProducer の send メソッドが呼び出される限り、成功したとみなされます。
all または -1: メッセージが送信されたとみなされ、成功した送信がクライアントに返される前に、リーダーノードがメッセージを保存しているだけでなく、そのすべてのレプリカ (正確には、ISR 内のノード) がメッセージを保存している必要があることを示します。これは最も厳格な永続性保証ですが、当然ながらパフォーマンスは最も低くなります。
1: 送信成功がクライアントに返される前に、メッセージをリーダーノードに書き込むだけでよいことを示します。

再試行パラメータの役割

運用側で Kafka によって提供されるもう 1 つのコアプロパティは、メッセージの送信が失敗した後の再試行回数を制御するために使用されます。 0 に設定すると再試行は行われません。再試行すると、送信側でメッセージが重複する可能性があります。メッセージ送信インターフェースの観点から:

 Future<RecordMetadata> は ProducerRecord<K, V> レコードを送信します。 Future<RecordMetadata> を送信します(ProducerRecord<K, V> レコード、コールバック コールバック);

上記の API から、ユーザーが KafkaProducer を使用してメッセージを送信する場合、まず送信するメッセージを ProducerRecord にカプセル化して、典型的な Future 設計パターンである Future オブジェクトを返す必要があることがわかります。

Kafka メッセージ追加プロセス

KafkaProducer の send メソッドは、メッセージをブローカーに直接送信しません。 Kafka ではメッセージの送信が非同期になり、2 つのステップに分割されます。送信メソッドの役割は、メッセージをメモリ (パーティション化されたキャッシュキュー) に追加することであり、その後、専用の送信スレッドがキャッシュされたメッセージをバッチで Kafka ブローカーに非同期的に送信します。

メインメソッドはKafkaProducer#doSendです

メッセージをプロデューサーの送信バッファに追加します。その実装クラスは RecordAccumulator です。まず、メモリに書き込まれる Kafka メッセージのフローチャートを見てみましょう。

送信者スレッド

これまで、send メソッドを呼び出すと、実際にはプロデューサークライアントのサービスメモリにのみ送信されることを確認しました。まだブローカーに連絡が取れていません。 Kafka プロデューサークライアントのバックグラウンドでは、スレッドが開始され、メッセージバッチが保存されている領域が継続的にポーリングされ、ブローカーにメッセージが送信されます。

メッセージバッチのメモリ構造と割り当て

上記のソースコードから、各 ProducerBatch は batch.size バイトのメモリブロックであることがわかります。そしてプーリング技術が使われます。

バッファプールのメモリ保持クラスは BufferPool です。まず、BufferPool のメンバーを見てみましょう。

パブリッククラス BufferPool {
  //合計メモリサイズ
  プライベート最終長い合計メモリ;
  // 各メモリブロックのサイズ、つまりbatch.size  
  プライベート最終intプール可能なサイズ;
  // メモリの適用と返却のメソッドの同期ロック
  プライベート最終ReentrantLockロック;
  // メモリブロックを解放する
  プライベート最終 Deque<ByteBuffer>を解放します。
  // 空きメモリブロックのイベントを待つ必要がある
  プライベート最終 Deque<Condition> waiters;
  /** 利用可能なメモリの合計は  nonPooledAvailableMemoryとバイトバッファの数 空き* poolableSize。 */
  // バッファ プールにはまだ空きメモリが割り当てられていません。新しく適用されたメモリブロックはここからメモリ値を取得します
  プライベート long nonPooledAvailableMemory;
  // ...
 }

BufferPool のメンバーから、バッファープールが実際には ByteBuffers で構成されていることがわかります。 BufferPool はこれらのメモリブロックを保持し、メンバー free に保存します。 free の合計サイズは totalMemory によって制限され、nonPooledAvailableMemory はバッファープールに割り当てられていないメモリがどれだけ残っているかを示します。

バッチメッセージが送信されると、そのメッセージが保持するメモリブロックは free に戻されるため、後続のバッチがメモリブロックを適用したときに新しい ByteBuffer は作成されず、free から取得できるため、JVM によってメモリブロックが再利用される問題を回避できます。

メモリブロックを作成するプロセスは次のとおりです。

メモリブロックを返す論理フロー

返されたメモリブロックのサイズが batchSize と等しい場合、メモリブロックはクリアされ、バッファープールの空き領域に追加されます。つまり、メモリブロックを再利用するための JVM GC を回避して、メモリブロックがバッファープールに返されます。等しくない場合は、未割り当ておよび空きメモリサイズの値にメモリサイズを追加します。メモリを返す必要はなく、JVM GC がメモリをリサイクルするのを待ってから、空きメモリを待機しているスレッドを起動します。

Java プロデューサーは TCP 接続をどのように管理しますか?

なぜ TCP なのか?

Apache Kafka のすべての通信は、HTTP やその他のプロトコルではなく、TCP に基づいています。これは、生産者、消費者、ブローカー間の通信にも当てはまります。

コミュニティの観点から見ると、クライアントを開発する際に、多重化要求や複数の接続を同時にポーリングする機能など、TCP 自体が提供する高度な機能の一部を活用できます。

TCP の多重化要求により、物理接続上に複数の仮想接続が作成され、各仮想接続は対応する独自のデータストリームを送信する役割を担います。厳密に言えば、TCP は多重化できません。失われたメッセージの自動再送信など、信頼性の高いメッセージ配信セマンティクスのみを提供します。

さらに、現在知られている HTTP ライブラリは、多くのプログラミング言語ではやや初歩的なものです。

TCP 接続はいつ作成されますか?

KafkaProducer インスタンスが作成されると、TCP 接続が確立されます。 KafkaProducer インスタンスが作成されると、プロデューサーアプリケーションはバックグラウンドで Sender という名前のスレッドを作成して開始します。送信スレッドが実行を開始すると、最初にブローカーとの接続が作成されます。

プロパティ properties = new Properties();
 properties.put( "bootstrap.servers" , "localhost:9092" );
 properties.put( "key.serializer" 、 StringSerializer.class.getName());
 properties.put( "value.serializer" 、 StringSerializer.class.getName());
 //リソースを使って試す
// KafkaProducer インスタンスを作成すると、Sender スレッドが作成され、バックグラウンドで開始されます。送信スレッドが実行を開始すると、まずブローカーとのTCP接続を作成します。
試してください (Producer<String, String> producer = new KafkaProducer<>(properties)) {
    ProducerRecord<String, String> レコード = new ProducerRecord<>(TOPIC, KEY , VALUE);
    コールバック callback = (メタデータ、例外) -> {
    };
    プロデューサー.send(レコード、コールバック);
 }

bootstrap.serversはプロデューサーのコアパラメータの1つであり、プロデューサーが起動したときに接続するブローカーのアドレスを指定します。
bootstrap.servers で 1000 個のブローカーが指定されている場合、プロデューサーは起動時にまずこれらの 1000 個のブローカーとの TCP 接続を作成します。
したがって、クラスター内のすべての Broker 情報を bootstrap.servers に構成することはお勧めしません。通常は 3 ～ 4 台のサーバーで十分です。
プロデューサーがクラスター内の任意のブローカーに接続されると、クラスター全体のブローカー情報を取得できます（メタデータ要求）

TCP 接続は、メタデータの更新後とメッセージの送信時の 2 つの場所で作成される場合もあります。

プロデューサーがクラスターメタデータを更新するときに、一部のブローカーとの接続がないことが判明した場合、プロデューサーはTCP接続を作成します。

【シーン1】

プロデューサーが存在しないトピックにメッセージを送信しようとすると、ブローカーはトピックが存在しないことをプロデューサーに伝えます。このとき、プロデューサーは Kafka クラスターにメタデータ要求を送信し、最新のメタデータ情報を取得し、クラスター内のすべてのブローカーとの TCP 接続を確立しようとします。

【シーン2】

プロデューサーは、metadata.max.age.ms パラメータを通じてメタデータ情報を定期的に更新します。デフォルト値は 300000、つまり 5 分です。

プロデューサーがメッセージを送信しようとすると、プロデューサーはターゲットブローカーとの接続がないことを検出し (負荷分散アルゴリズムによって異なります)、TCP 接続も作成します。

TCP 接続はいつ閉じられますか?

プロデューサーが TCP 接続を閉じる方法は 2 つあります。ユーザーによるアクティブなクローズと Kafka による自動クローズです。

[ユーザーアクティブシャットダウン]

広い意味では、アクティブシャットダウンには、ユーザーが kill -9 を呼び出して Producer を強制終了することが含まれます。最も推奨される方法は、producer.close() です。

[Kafka は自動的にシャットダウンします]

プロデューサーパラメータ connections.max.idle.ms のデフォルト値は 540000 (9 分) です。

9 分以内に TCP 接続を介してリクエストが渡されない場合、Kafka は TCP 接続をアクティブに閉じます。

connections.max.idle.ms=-1 はこのメカニズムを無効にし、TCP 接続は永続的な長時間接続になります。

Kafka によって作成されたソケット接続ではすべてキープアライブが有効になっています。

【知らせ】

TCP 接続を閉じるイニシエーターは Kafka クライアントであり、これは受動的な閉じるシナリオです。

受動的なクローズの結果、多数のCLOSE_WAIT接続が生成される。

プロデューサーまたはクライアントには、TCP 接続が切断されたことを明示的に確認する機会がありません。

要約する

これで、冒頭の 3 つの質問に答えることができます。

1. Kafka はメッセージを個別に送信しますか、それともバッチで送信しますか?

通常は一括して送信されます。 ProducerBatch にカプセル化して送信します。

2. Kafka はどのようにして単一のメッセージを送信しますか?

send メソッドを同期的に呼び出すには、単一のプロデューサーと単一のスレッドのみを設定できます。

3. Kafka はメッセージを順番に送信しますか?

いいえ、順序が必要な場合は、キーを設定する必要があり、プロデューサーはシングルスレッドになります。

4. プロデューサーが FullGC を頻繁に実行するのはどのような状況でしょうか?

メッセージサイズが batchSize より大きい場合、割り当てられたメモリブロックはループ内で free から取得されませんが、新しい ByteBuffer が作成され、ByteBuffer はバッファープールに返されません (JVM GC リカバリ)。この時点で nonPooledAvailableMemory がメッセージ本文より小さい場合、free 内の空きメモリブロックが破棄され (JVM GC リカバリ)、ユーザーアプリケーション用のバッファプールに十分なメモリ領域が確保されます。これらのアクションにより、GC の問題が頻繁に発生します。

したがって、頻繁な GC を回避するには、ビジネスメッセージのサイズに応じて batch.size を適切に調整する必要があります。

<<: GitOps 継続的デプロイメントツールである Argo CD を初めて体験

>>: 最高のPython仮想環境。