大企業から採用されたアーキテクトは、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 ミリ秒後に強制的に送信される可能性があります。これは良い考えではありません。バッチが役に立たなくなり、十分なデータが収集されなくなります。

6. 最大リクエストサイズ

パラメータ「max.request.size」は、Kafka サーバーに送信される各リクエストの最大サイズを決定します。また、メッセージの最大サイズもこのパラメータで設定された値に制限されます。これは実際には、独自のメッセージのサイズに応じて柔軟に調整できます。

例を挙げてみましょう。御社から送信されるメッセージはすべて大きなテキストメッセージです。各メッセージには大量のデータが含まれています。 1 つのメッセージは 20 KB になる場合があります。

この時点で、batch.size をより大きなサイズに調整する必要がありますか?例えば512KBに設定しますか？では、より大きな buffer.memory を与えるべきでしょうか?例えば128MBに設定しますか？

この方法でのみ、バッチ メカニズムを使用して、大規模なメッセージ シナリオで複数のメッセージをパッケージ化できます。しかし、この時点で「max.request.size」も同期的に増やす必要があるのでしょうか?

おそらく、リクエストの 1 つが非常に大きいためです。デフォルトでは1MBです。適切に、例えば 5MB に増やすことはできますか?

7. 再試行メカニズム

「retries」と「retries.backoff.ms」は再試行メカニズム、つまり、リクエストが失敗した場合に何回再試行できるか、また各再試行の間隔を何ミリ秒にするかを決定します。

このため、いくつかの再試行機会を適切に設定し、100 ミリ秒の再試行間隔など、特定の再試行間隔を指定できます。

8. 持続メカニズム

「acks」パラメータは、送信されたメッセージに使用される永続化戦略を決定します。これには、他の多くの概念が含まれます。