大企業から採用されたアーキテクトは、Kafka パラメータのチューニングを非常にエレガントにこなしました。たくさんのことを学びました。

1. 背景紹介: 多くの学生はKafkaパラメータを理解していない

今日は非常に興味深い話題についてお話ししたいと思います。ご存知のとおり、多くの企業が MQ として Kafka をベースにした複雑な大規模システムを開発しています。

Kafka クライアントを使用してサーバーと対話するコードを記述する場合、クライアントに多くのパラメータを設定する必要があります。

そのため、チームに参加したばかりで Kafka テクノロジーについてあまり知らない若いクラスメートにたくさん会いました。

この時点で、彼らはチーム内の上級の同僚が書いたコードを見て、何が起こっているのか、その背後にある意味、特にいくつかのKafka パラメータ設定を理解していないでしょう。

そのため、この記事では、Kafka クライアントによって設定されたいくつかのパラメータを次に見たときに怖がらないように、図を描くという古いルーチンを使用して、Kafka プロダクション側でのいくつかの一般的なパラメータの設定について説明します。

2. Kafkaプロダクション終了時のサンプルコード

 プロパティprops = new Properties ();
 小道具。 ( "bootstrap.servers" 、 "localhost:9092" ) を配置します。
 小道具。 ( "key.serializer" 、 "org.apache.kafka.common.serialization.StringSerializer" ) を配置します。
 小道具。 ( "value.serializer" 、 "org.apache.kafka.common.serialization.StringSerializer" ) を配置します。
 小道具。 put ( "buffer.memory" 、 67108864 );
 小道具。 put ( "batch.size" 、 131072 );
 小道具。 ( "linger.ms" 、 100 ) を入力します。
 小道具。 ( "最大リクエストサイズ" 、 10485760 ) を設定します。
 小道具。 ( "acks" 、 "1" ) を置く。
 小道具。 put ( "再試行" 、 10 );
 小道具。 ( "retry.backoff.ms" 、 500 ) を設定します。

 KafkaProducer < String , String > プロデューサー= 新しいKafkaProducer < String , String > ( props );

3. メモリバッファサイズ

まず、「buffer.memory」というパラメータが何を意味するのか見てみましょう。

Kafka クライアントがデータをサーバーに送信する場合、通常はバッファリングを経由する必要があります。つまり、KafkaProducer を介して送信するメッセージは、最初にクライアントのローカルメモリバッファーに入り、その後、多数のメッセージがバッチに集められて Broker に送信されます。

したがって、この「buffer.memory」の本質は、KafkaProducer が使用できるメモリバッファーのサイズを制限することです。デフォルト値は 32MB です。

意味がわかったところで、本番プロジェクトでこのパラメータを設定する方法を考えてみましょう。

まず、メモリバッファの設定が小さすぎるとどのような問題が発生する可能性があるかを考えるかもしれません。

まず、多数のメッセージがメモリバッファーにバッファーされ、それぞれに複数のメッセージが含まれるバッチが形成されることを明確にする必要があります。

次に、KafkaProducer には、複数のバッチをリクエストにパッケージ化して Kafka サーバーに送信する Sender スレッドがあります。

メモリの設定が小さすぎると、メッセージはすぐにメモリバッファーに書き込まれますが、送信スレッドには Kafka サーバーにリクエストを送信する時間がないという問題が発生する可能性があります。

これにより、メモリバッファがすぐにいっぱいになりますか?いっぱいになると、ユーザースレッドがブロックされ、それ以上のメッセージは Kafka に書き込まれなくなります。

したがって、実際の状況に基づいて、「buffer.memory」パラメータのストレステストを実行する必要があります。実稼働環境では、ユーザースレッドがメモリバッファーに 1 秒あたりに書き込むメッセージの数を計算する必要があります。

たとえば、1 秒あたり 300 件のメッセージがある場合は、ストレステストを実行する必要があります。メモリバッファーが 32 MB で、1 秒あたり 300 件のメッセージがメモリバッファーに書き込まれると仮定すると、メモリバッファーは頻繁にいっぱいになりますか?このようなストレステストを行った後、適切なメモリサイズをデバッグできます。

4. バッチにパッケージ化する必要があるデータの量はどれくらいですか?

次に、2 番目の質問である「batch.size」をどのように設定するかについて考える必要があります。これにより、送信前に各バッチに保存する必要があるデータの量が決まります。

たとえば、バッチのサイズを 16 KB に設定すると、バッチに 16 KB のデータがあれば送信できます。

このパラメータのデフォルト値は 16KB です。通常は、このパラメータをより大きな値に調整し、独自の運用環境でメッセージを送信する負荷を使用してテストすることができます。

たとえば、メッセージの送信頻度が 1 秒あたり 300 の場合、「batch.size」を 32KB または 64KB に調整すると、メッセージ送信の全体的なスループットが向上しますか?

理論的には、バッチサイズを大きくすると、より多くのデータをバッファリングできるため、1 回のリクエストで送信されるデータの量が増え、スループットが向上する可能性があります。

しかし、この物体は無限に大きくなることはできません。大きすぎる場合、データが常にバッチ内にバッファリングされ、長時間送信されないと、メッセージ送信の遅延が非常に大きくなります。

たとえば、メッセージがバッチに入ると、バッチが 64 KB でいっぱいになってから送信されるまでに 5 秒かかります。このメッセージの遅延は 5 秒です。

したがって、本番環境のメッセージ送信速度に応じてさまざまなバッチサイズを調整し、最終的なスループットとメッセージ遅延を自分でテストして、最も合理的なパラメーターを設定する必要があります。

5. バッチを長時間満たすことができない場合はどうなりますか?

バッチを長時間満たすことができない場合は、別のパラメータ「linger.ms」を導入する必要があります。

つまり、バッチが作成されると、バッチがいっぱいかどうかに関係なく、バッチを送信する必要があるということです。

例を挙げてみましょう。たとえば、batch.size は 16kb ですが、ピーク時以外はメッセージの送信が非常に遅くなります。

これにより、バッチの作成後にメッセージが次々に届くようになりますが、16KB を蓄積するには長い時間がかかります。現時点ではただ待つしかないのでしょうか?

もちろん違います。「linger.ms」を 50ms に設定したとします。すると、バッチの作成から 50 ミリ秒が経過していれば、16 KB いっぱいでなくても送信されます。

したがって、「linger.ms」は、メッセージがバッチに書き込まれると、最大でこの時間待機し、その後バッチとともに送信されることを決定します。

バッチを完全に埋めることができず、メモリ内にメッセージがバックログされて送信できなくなる状況を回避します。これは非常に重要なパラメータです。

このパラメータは通常、非常に慎重に設定する必要があり、batch.size と一緒に設定する必要があります。

たとえば、最初にバッチが 32 KB であると仮定し、通常の状況でバッチを完了するのにどのくらいの時間がかかるかを見積もる必要があります。たとえば、通常の状況ではバッチを完了するのに 20 ミリ秒かかる場合があります。

次に、linger.ms を 25ms に設定します。つまり、通常はほとんどのバッチは 20 ミリ秒以内に満たされますが、linger.ms を使用すると、オフピーク期間中であってもバッチが 20 ミリ秒以内に満たされない場合でも、バッチは 25 ミリ秒後に強制的に送信されるようになります。

linger.ms を小さく設定しすぎると (たとえば、デフォルトは 0 ミリ秒ですが、これを 5 ミリ秒に設定すると)、バッチが 32 KB に設定されているにもかかわらず、32 KB を収集するのに十分なデータがない場合が多く、バッチは 5 ミリ秒後に強制的に送信される可能性があります。これは良い考えではありません。バッチが役に立たなくなり、十分なデータが収集されなくなります。