Kafka エキスパートになるためのヒント: プロデューサーの詳細な実践概要

著者について

Zhang Jinwei 氏は、Tencent Cloud のメッセージ キュー特別サポート チームのメンバーであり、Kafka と Puslar のシニア デベロッパーであり、Kafka SDK の貢献者でもあります。彼はストリーミング データ処理とメッセージ キューの分野で長年の実務経験を持っています。

Kafka の紹介

Kafka は、リリースされてからほぼ 10 年になる分散メッセージ キュー システムです。大手企業や大手製品に多かれ少なかれ採用されている、非常に成熟した製品です。特に、ビッグデータ ストリーム処理やログ ストリーム処理などのシナリオでは、Kafka はほぼ不可欠な役割を果たします。

このメッセージ キューは、正式には「分散ストリーム処理プラットフォーム」として定義されています。その主な目的は、大規模なデータ ストリーム処理でレコード ストリームを保存することです。しかし、この時代では、より重くパフォーマンスが制限されている RabbitMQ の代わりに、メッセージ キューとして Kafka を使用することを好むビジネス アーキテクチャが増えています。

Kafka のようなシステムのシンプルさと高いパフォーマンスを確保するために、多くの論理的な詳細と構成が実際にはクライアント側に配置されます。したがって、ここではクライアントの視点から始め、プロデューサーとコンシューマーの両方を介したユーザーの視点から、実際の運用で Kafka が遭遇するいくつかの問題と対応する技術的な詳細を紹介します。この記事はプロデューサーを紹介するシリーズの第1回目です。

標準プロデューサー API の紹介

ここではまず、最もよく使用されるプロデューサー API を紹介します。この記事を読んだら、あなたはすでに Kafka についてよく知っていると思います。ただし、後で使用される可能性のある用語をいくつか紹介するために、まず Kafka の基本的な概念を確認しましょう。ここでは、制作部分にのみ焦点を当て、その他の無関係な詳細は無視します。

まず、プロデューサーと Kafka 間の相互作用の図を描きます。この図は、プロデューサー メッセージ データのフローと、Kafka サーバーがメッセージを受信するために使用する必要があるコンポーネントを説明するために使用されます。

写真は上の通りです。それでは、図に描かれている内容とそれに対応する専門用語をそれぞれ紹介しましょう。

• クライアントとは、メッセージを書き込む複数の異なるクライアントを指します。ここでのクライアントは抽象的な概念です。 Kafka サーバーとの接続が確立されている限り、メッセージは Kafka に書き込まれます。同じサーバー内にあるか、同じプロセス内にあるかに関係なく、クライアントと呼びます。
• ブローカーとは、クラスターに追加されたサーバー、つまり物理レベルのマシン ノードを指します。 Kafka がマシンにデプロイされ、Kafka クラスターに追加されると、このマシンは Kafka クラスターのノードになります。通常、マシンには 1 つの Kafka サービスのみがデプロイされます。
• トピックは抽象的な概念です。その主な機能は、ユーザーが処理したメッセージをさまざまなカテゴリに分類することです。同時に、各トピックには、最大メッセージ サイズなどのトピック ディメンションの構成がいくつかある場合があります。実際にメッセージを伝送するために、トピックの下にさまざまな数のパーティションが作成されます。ここで注目すべきは、Kafka が RabbitMQ とは異なるということです。 Kafka によって作成されたトピックごとに、対応する数のパーティションがブローカー上に作成されます。したがって、Kafka のトピックの数は限られているので、あまり多くしないようにしてください。
• RabbitMQ のトピックはルーティングの概念です。非常に多くのトピックを作成しても、実際にそれらを運ぶキューは作成されず、サブスクライブおよび配信時に異なるルーティング戦略が実行されるだけです。これは、RabbitMQ から Kafka に切り替える多くのユーザーに共通する問題です。
• パーティションは、メッセージを伝送するために使用されるトピック パーティションです。パーティション間に相関関係はありません。それぞれに独自の順序、メッセージの内容、記録されたオフセットがあります。各プロデューサーのメッセージはパーティションに書き込まれ、プロデューサーは独自のアルゴリズムに従って書き込むパーティションを選択します。
• レプリカはパーティションのコピーであり、物理レベルと概念レベルの両方で最小の単位です。彼らはブローカーに縛られます。各パーティションには、メッセージが実際に書き込まれるレプリカが少なくとも 1 つ必要です。パーティションに複数のレプリカがある場合、コントローラーはどのレプリカがリーダーになるかを決定します。メッセージは常にリーダーに書き込まれ、その後リーダーはフォロワー状態にある他のレプリカにデータを同期します。したがって、上の図に示すように、クライアントのメッセージがパーティションに書き込まれるように選択された後、実際には、クライアントはレプリカ リーダーが配置されているブローカーに接続し、そこにメッセージを書き込みます。

ここで、プロデューサー API の使用と、制作中に注意する必要があるいくつかの構成について説明します。ここでのプロデューサー API は、Kafka によって提供されるほぼステートレスな API を指し、非常に軽量で、優れたパフォーマンスも提供できます。ただし、この API を本番環境で使用する場合、Kafka は少なくとも 1 回と最大で 1 回のセマンティクスのみを保証します。

• 「少なくとも 1 回」は、メッセージが繰り返される可能性があることを意味します。メッセージが重複している場合、コンシューマーは消費時に重複を排除する必要があります。
• 「At-most-once」はほとんど使用されません。これは通常、大量の重要でないデータを処理するときに使用され、データ損失を許容しながら比較的良好なパフォーマンスを提供できます。生成側がメッセージを生成して配信した後は、それが成功したかどうかは考慮されず、再試行も行われません。
• 現在のプロデューサー API では、正確に 1 回のセマンティクスは提供されていません。ただし、多数のストリーミング コンピューティング システムでは正確に 1 回のセマンティクスを保証する必要があり、Kafka も Kafka Stream などのストリーミング処理フレームワークを開始しているため、Kafka の以降の新しいバージョンでは、正確に 1 回のセマンティクスを保証するトランザクション メッセージが提供されます。この問題については、この記事の次の章で説明します。

次に、コード スニペットを使用して、プロデューサー API の使用方法を確認し、いくつかの重要な構成を確認します。まず、それぞれに 2 つのレプリカがある 2 つのパーティションを持つトピックを作成しましょう。

 bin/kafka-topics.sh --create --bootstrap-server localhost: 9092 --replication-factor 2 --partitions 2 --topic test

それでは、Java プロデューサー スニペットを記述してみましょう。コードは非常にシンプルです。プロデューサー クライアント関連の構成をいくつか構成し、プロデューサーの send メソッドを呼び出して、Kafka クライアント ライブラリに送信するメッセージを送信するだけです。

プロパティ props = new Properties(); 
 
 props.put( "bootstrap.servers" , "localhost:9092" ); 
 
 props.put( "acks" , "all" ); 
 
 props.put( "再試行" , 3 ); 
 
 props.put( "retry.backoff.ms" , 2000 ); 
 
 props.put( "圧縮タイプ" , "lz4" ); 
 
 props.put( "バッチサイズ" , 16384 ); 
 
 props.put( "linger.ms" , 200 ); 
 
 props.put( "最大リクエストサイズ" , 1048576 ); 
 
 props.put( "key.serializer" 、 "org.apache.kafka.common.serialization.StringSerializer" ); 
 
 props.put( "value.serializer" 、 "org.apache.kafka.common.serialization.StringSerializer" ); 
 
 props.put( "リクエストタイムアウト" , 10000 ); 
 
 props.put( "max.block.ms" , 30000 ); 
 
プロデューサー<String, String>プロデューサー =新しいKafkaProducer<String, String>(props); ( int i = 0 ; i < 1000 ; i++) { 
 
 Future<RecordMetadata> future = producer.send( new ProducerRecord<>( "test" , UUID.randomUUID().toString())); 
 
 System.out.println( "オフセットを生成します:" + future.get().offset()); 
 
 } 
 
プロデューサー.close();

上記のコードでメッセージの生成を完了できます。 Kafka が提供する API は実にシンプルで使いやすいです。もちろん、実際にはさらに詳細な情報が含まれていますが、それらはクライアントによってカプセル化されています。ここではさらに深く掘り下げて、このコードに潜む潜在的な落とし穴を確認します。

まず、このコードでクライアントが何を行うかを簡単に分析してみましょう。 (注: ここでは一般的なワークフローを示すことを優先しているため、Java クライアントの一部の固有機能は無視される場合があります)。

1. クライアントは、コードで指定された bootstrap.servers を介してブローカーに接続します。最初のブローカー接続が失敗した場合、クライアントは、すべての接続が失敗するか、1 つのアドレスが正常に接続されるまで、左から右へ接続を再試行します。

2. 接続が成功すると、Kafka クライアントは Kafka サーバーに ApiVersions リクエストを開始し、サポートされている各 API と各 API の最大サポートバージョンをサーバーに照会します。これにより、Kafka の下位互換性の目的が達成されます。もちろん、ApiVersions はバージョン 0.10 あたりで追加された API なので、新しいバージョンのクライアントが 0.9 バージョンの kafka サーバーにアクセスすると、ArrayIndexOutOfBoundsException エラーが発生します。このエラーは公式 Kafka 0.10 で修正されました。

3. 次に、クライアントは、メッセージが送信されるトピックのメタデータ情報を照会し、接続されたブローカーにメタデータ要求を送信します。この API を通じて、Kafka クライアントは、IP やポート、ブローカーに対応する一意の ID など、クラスター ブローカーに関するさまざまな情報を取得します。同時に、クライアントはトピック、パーティション ID、およびパーティションによって選択されたリーダー レプリカが配置されているブローカーの ID に関する関連情報も取得します。次に、クライアントは、実際にメッセージを送信するためのデータ リンクとして、リーダー レプリカが配置されているブローカーとの接続を確立します。

ここで注目すべき3つの点があります

• まず、Kafka サーバーが auto.create.topics.enable を true に設定して構成されている場合、クライアントが存在しないトピックのメタデータを照会すると、トピックは Kafka サーバーによって自動的に作成されます。
• 2 番目に、一般的に言えば、Kafka クライアントによるメタデータの処理は定期的な更新アクションです。メタデータが 30 秒ごとに更新される場合、この 30 秒間にユーザーがトピック構成を変更してパーティションを追加しても、クライアントはそれを認識できません。トピックに追加のパーティションがあることを本番側が認識し、新しいパーティションにデータを書き込むことができるようになるには、クライアントがメタデータを更新するまで待機する必要があります。
• メタデータを更新するときに、クライアントの運用トラフィックが Kafka のクォータ制限を超え続け、Kafka のスロットリングが発生すると、メタデータ データの取得が再試行され、タイムアウトになります。この極端なケースでは、クライアントはパーティションの追加を非常に長い間認識できない可能性があります。
• こうした状況は、生産現場でも発生しています。使用中に、本番側で新しく作成されたパーティションにメッセージを書き込むのが遅い場合は、おそらくこの方向から始めることができます。
• 3 番目に、Kafka の接続確立ロジックの観点から見ると、Kafka は実際には複数のリンクを確立し、クラスター内の異なるブローカーに直接接続します。したがって、ここでファイアウォールポリシーを適用する場合は、ブローカーごとにポリシーを適用する必要があります。そうしないと、メタデータを取得できても、メッセージを生成できない可能性があります。

4. クライアントは、メッセージ内のキーに基づいてハッシュを計算し、メッセージが配信されるパーティションを決定します。次に、クライアントはメッセージをローカル キューに配信します。パーティションに実際に接続されている配信クラスがキューからメッセージを取得します。次に、クライアントは 2 つのチェックを実行し、メッセージを配信するために生成要求を呼び出します。

• メッセージがmax.request.sizeより大きい場合、RecordTooLargeExceptionが直接返されます。
• メッセージがbatch.sizeより小さい場合は、合計メッセージサイズがbatch.sizeより大きくなるか、linger.msで指定された時間を超えるまで、後続のメッセージを待機します。
• クライアントは、非同期または同期プロセスを開始して生成要求の戻りを待機し、構成された再試行戦略に基づいて再試行を実行するか、失敗した送信のエラーをビジネス ロジックに返して、ビジネス ロジックがエラーを処理できるようにします。

5. このステップでは主に以下の設定を行います。

• acks は -1 または all です。これは、メッセージが正常に書き込まれた後にのみ、isr (同期中) リスト内のすべてのレプリカがクライアントに成功を返すことを意味します。同時に、トピック レベルで min.insync.replicas 構成も提供されます。 ISR 内のレプリカ数がこの構成の値より少ない場合も、書き込みは失敗します。これは Kafka が提供できる最も強力な制約です。
• acks は 0 です。つまり、メッセージはクライアントの TCP ソケット バッファーに配信されるとすぐに送信されたと見なされ、クライアントは Kafka クラスターがメッセージを受信したか、正常に書き込んだかどうかを気にする必要がなくなります。このモードでは、Kafka は最大 1 回のセマンティクスのみを提供します。これは、データ損失が許容される場合に最高のパフォーマンス モードです。
• acks は 1 です。つまり、メッセージがレプリカ リーダーに送信され、正常に書き込まれた場合、他のフォロワーが正常に書き込んだかどうかに関係なく、成功が返されます。メッセージが配信された直後にリーダーが電話を切ると、メッセージは失われます。
• このモードでは、ほとんどの場合メッセージが失われないことが保証され、パフォーマンスとセキュリティのバランスが取れたモードです。
• サーバーは、クライアントによって提供されるacks構成値に基づいて、どのような状況でサーバーの書き込みがクライアントに返されるかを決定します。
• クライアントが書き込み失敗を受信した場合、再試行の間に retry.backoff.ms で定義された時間を置いて、retries で設定された回数だけ再試行します。再試行回数が尽きた後にのみ、失敗がビジネス ロジックに返されます。

上記は、使用中のプロデューサー API 全体の詳細です。これらの詳細を理解すれば、運用中に Kafka で発生した奇妙なエラーを見つけて処理するためのアイデアが得られるでしょう。もちろん、コードの観点から見ると、上記の記事で説明されていないコード内の構成がいくつかあります。ここでは一緒に紹介します

• compression.type は圧縮を設定するために使用されます。 Kafka は、none (圧縮なし)、gzip、snappy、lz4、zstd (Kafka バージョン 2.1.0 以降が必要) など、さまざまな圧縮モードを提供します。
• 一般的に、比較的軽い CPU 負荷で良好な圧縮率と非常に高いスループットを提供できる lz4 圧縮をお勧めします。全体的なパフォーマンスとコストパフォーマンスは、他のいくつかの圧縮方法よりも優れています。したがって、圧縮率に対する強い要求がない場合は、lz4 の方が適しています。
• key.serializer と value.serializer は、Java クライアントに固有のシリアライザーであり、キーと値のシリアル化方法を決定するために使用されます。ここでは詳細には触れません。
• request.timeout.ms 構成は、ネットワーク要求のタイムアウトを定義します。このパラメータで指定された時間内に Kafka クライアントがサーバーへの応答を受信しない場合、リクエストはキャンセルされ、失敗したとみなされ、ロジックは失敗処理ロジックに転送されます。この制約は比較的強い制約です。
• max.block.ms この構成項目は、クライアント内のブロック時間を定義します。たとえば、内部の非同期キューがいっぱいの場合、Kafka クライアントは、send を呼び出すときにタイムアウトになるまで待機します。このパラメータに関して注意すべき点は、ユーザー定義のシリアライザーとパーティショナーで費やされた時間は、このパラメータのタイムアウトにはカウントされないことです。

冪等プロデューサーの紹介

べき等プロデューサーは、単一のパーティション上のプロデューサーに対して正確に 1 回のセマンティクスを提供しますが、複数のパーティション上のプロデューサーの操作の一貫性をカバーすることはできません。この一貫性は、後続のトランザクション メッセージを通じて解決する必要があります。

それでは、べき等プロデューサーの使用方法と、関連する詳細の一部を見てみましょう。

なぜべき等プロデューサーを使用する必要があるのでしょうか?主な理由は、プロデューサーがサーバーにメッセージを送信した後、ネットワークの問題によって接続が切断された場合、プロデューサーはメッセージが正常に書き込まれたか失敗したかを認識できないことです。 Kafka の一般的なプロデューサー API では、再試行パラメータを設定し、常に再試行します。書き込みが成功したかどうかを認識できないため、これを「少なくとも 1 回」セマンティクスと呼びます。書き込みが成功したが、成功した応答を受信しなかった場合は再試行し、メッセージが繰り返し書き込まれることになります。メッセージの消費がメッセージの順序に依存する場合、この再試行によって順序が乱れる可能性もあります。

現在、べき等プロデューサーを使用すると、Kafka は再試行を実行するときにそのような重複したメッセージを破棄できます。

次に、べき等プロデューサーの使い方を見てみましょう。 (ここでコードを見てみましょう。コード内の重要でない構成は無視します)。

プロパティ props = new Properties(); 
 
 props.put( "bootstrap.servers" , "localhost:9092" ); 
 
 props.put( "enable.idempotence" , true ); 
 
プロデューサー<String, String>プロデューサー =新しいKafkaProducer<String, String>(props); ( int i = 0 ; i < 1000 ; i++) { 
 
 Future<RecordMetadata> future = producer.send( new ProducerRecord<>( "test" , UUID.randomUUID().toString())); 
 
 System.out.println( "オフセットを生成します:" + future.get().offset()); 
 
 } 
 
プロデューサー.close();

ユーザーの使用状況の観点から、べき等プロデューサーを有効にするには、enable.idempotence 設定を true に追加するだけです。引き続き詳細に注意を払い、べき等プロデューサーを有効にした後に kafka クライアントが何を行うかを確認しましょう。まず、クライアントはプロデューサーのいくつかの構成値を強制します。

1. ack は強制的にすべてに設定されます。顧客が元々 acks レベル 0 または 1 を使用している場合、ユーザーはそれをすべてに設定することによるビジネス パフォーマンスへの影響を考慮する必要があります。ユーザーが最初にこれを all に設定した場合、べき等プロデューサーを使用しても追加コストはほとんど発生しません。
2. 再試行回数は 1 より大きい数値に設定する必要があります。通常、librdkafka および java kafka クライアントは、再試行回数を Integer.MAX_VALUE などの非常に大きな数値 (無限再試行に近い数値) に設定します。メッセージが正常に送信されることを確認します。
3. max.inflight.requests.per.connection は 5 未満である必要があります。Java Kafka クライアントのバージョンが 1.0.0 未満の場合、データ リンク上のリクエストが一度に 1 つだけになるように max.inflight.requests.per.connection が 1 に設定され、場合によっては TPS が低下します。
4. 送信されるメッセージ形式は v2 形式である必要があります。下位バージョンのメッセージ形式はサポートされていません。

プロデューサーの構成が完了すると、クライアントはプロダクション メッセージの送信を開始します。ここでは、前述の本番 API のロジックを省略し、冪等性を有効にした後の追加ロジックと手順のみに焦点を当てます。

1. プロデューサー クライアントは、ブローカーに対して InitProducerId リクエストを開始し、PID を要求します。送信される後続のすべてのメッセージには、プロデューサーを識別するためにこの PID が含まれます。
2. 各メッセージには単調に増加するシーケンス ID が付きます。 kafka サーバーは、同じ PID で送信された最後のメッセージのシーケンス ID を記録します。現在送信されているメッセージのシーケンス ID が最後に送信された ID 以下の場合、サーバーは現在のメッセージが期限切れであると見なし、メッセージの受信を拒否します。このような拒否要求を受信すると、クライアントは以前のメッセージが正常に配信されたことを認識し、送信の再試行を停止してメッセージを破棄します。

上記の手順により、Kafka は単一パーティション操作の各コンシューマーの冪等性を保証します。これは、特にコンシューマー API を使用しており、acks がすでに all に設定されている場合に非常に実用的な機能です。プロデューサーの冪等性を有効にしても、追加コストはほとんどかかりません。これも Kafka が長らくリリースしてきた機能 (Kafka 0.11 以降でサポート) ですが、この機能を使用しているユーザーはまだ比較的少ないようです。

トランザクションメッセージングの概要

トランザクション メッセージは現在、1 回限りのセマンティクスを保証するために Kafka によって提供される最も強力な制約です。これにより、プロデューサーが異なるパーティションに複数の相互に関連するメッセージを生成する場合、それらのメッセージが同時に成功するか、同時に失敗することが保証されます。同時に、トランザクション メッセージを有効にする前提として、べき等プロデューサーを有効にする必要があるため、べき等機能によって単一パーティション上の正確に 1 回のセマンティクスが保証されます。

ただし、一般的に、Kafka のトランザクション メッセージを直接使用する企業はほとんどありません。トランザクション メッセージの使用を伴うビジネスは、基本的に Kafka ストリームを介したストリーム処理です。 Kafka ストリームはトランザクション メッセージに依存しており、トランザクションの詳細を企業から隠します。ここでは、トランザクション メッセージを直接使用する方法を確認し、この期間中にクライアントが何を行うかを分析し続けます。まずはコードスニペットを公開しましょう。

プロパティ props = new Properties(); 
 
 props.put( "bootstrap.servers" , "localhost:9092" ); 
 
 props.put( "enable.idempotence" , "true" ); 
 
 props.put( "transactional.id" , "testtrans-1" ); 
 
 KafkaProducer<String, String> プロデューサー = new KafkaProducer(props); 
 
プロデューサー.initTransactions();試す{ 
 
プロデューサー.beginTransaction(); 
 
プロデューサー.send(レコード0); 
 
プロデューサー.send(レコード1); 
 
プロデューサー.sendOffsetsToTxn(…); 
 
プロデューサー.commitTransaction(); 
 
 }キャッチ(ProducerFencedException e) { 
 
プロデューサー.close(); 
 
 }キャッチ(KafkaException e) { 
 
プロデューサー.abortTransaction(); 
 
 }

まず、最初のステップとしてコード内で initTransactions が実行されます。このロジックでは、クライアントは InitProducerId を要求し、トランザクション ID を渡して、トランザクション ID と PID の間に 1 対 1 の関係を確立します。複数のプロデューサーが同じトランザクション ID に参加する場合、先に参加したプロデューサーは後に参加したプロデューサーに置き換えられます。以前のプロデューサーのリクエストはすべて拒否されます。

クライアントが短時間後に再接続する場合、InitProducerId を要求するときに、以前使用した PID とエポックを送信することに注意してください。成功した場合、サーバーはエポック+1 を返し、エポックが現在のエポックより小さいすべてのプロデューサー メッセージを拒否します。これは、分散システムにおけるいわゆるゾンビ問題を解決するためのものです。

その後のコード呼び出しは、多くのトランザクション コードと同じになります。トランザクションを開始し、書き込む必要のあるすべての情報を書き込み、最後にコミットします。失敗した場合はロールバックします。成功した場合、すべての書き込みがまとめてコミットされ、次のビジネス ロジックが実行されます。一般に、ほとんどのトランザクションはステート マシンとして実装されます。ここでは写真を載せて分析を続けません。

トランザクション コードを読んだところ、sendOffsetsToTxn 関数について言及していないようです。この関数は、現在のトランザクション メッセージがトピックから消費され、トランザクション メッセージに書き込まれるときに実際に使用されます。消費されたオフセットは、この機能を通じてコーディネータに送信され、その後、トランザクションがコミットされるときに、コンシューマーによって消費されたオフセットが一緒に送信されます。トランザクションの失敗を防ぐために、ユーザーはコンシューマー オフセットを手動で管理する必要もあります。非常に便利なヘルパー関数です。

要約する

これまで、Kafka プロデューサーのいくつかの使用法と、クライアントの観点から注意する必要がある落とし穴について簡単に分析しました。著者の記事は仕事で遭遇したいくつかの問題に基づいているので、いくつかの場所が使用され、より多くの人々がそれらを参考にするならば、もう少し詳しく書かれるかもしれません。問い合わせが少なく、問題も少ない場所であれば、よりシンプルな書き方で記述できるかもしれません。

この記事が読者の皆さんの役に立ち、Kafka プロデューサー部分についての理解を深めていただければ幸いです。読んでいただきありがとうございます。 Kafka コンシューマーについては次の記事でお会いしましょう。