Kafka のプロデューサーとコンシューマーのメカニズム + パーティショニング戦略、これ理解できないんですか?

[[442535]]

この記事はWeChat公式アカウント「ポスト00年代プログラマーXiaosan」から転載したもので、著者は003です。この記事を転載する場合は、2000年代生まれのプログラマーXiaosanの公式アカウントにご連絡ください。

カフカとは何か

Kafka はもともと Linkedin によって開発され、2010 年に Apache Foundation に寄贈され、トップクラスのオープンソース プロジェクトになりました。これは、Scala と Java で記述されたオープンソースの [分散ストリーム処理プラットフォーム] でもあります (MQ システムとも呼ばれますが、純粋なメッセージング システムではありません)。

現在、Kafka は分散ストリーム処理プラットフォームとして位置付けられています。高いスループット、永続性、水平スケーラビリティ、ストリーム データ処理のサポートなどの機能により、広く使用されています。現在、Cloudera、Storm、Spark、Flinkなどのオープンソースの分散処理システムがKafkaとの統合をサポートし始めています。

生産者と消費者のメカニズム

Kafka では、プロデューサーはブローカーにメッセージを送信し、ブローカーはプロデューサーから送信されたメッセージをディスクに保存します。コンシューマーはブローカーからのメッセージをサブスクライブして消費する責任があります。コンシューマーはプル モードを使用してサーバーからメッセージをプルします。 Zookeeper は、クラスター全体のメタデータ管理とコントローラーの選択を担当します。詳細は以下の図に示されています。

Kafka プロデューサープロデューサーはブローカーパーティション戦略を送信します

公開および購読の対象はトピックです。プロデューサーは指定されたトピックにメッセージを送信し、コンシューマーはサブスクライブされたトピックを消費します。 Kafka のパーティショニング メカニズムとは何ですか?トピックを複数のパーティションに分割し、各パーティションには複数のコピーがあり、同じトピックの下にある異なるパーティション内のメッセージも異なります。プロデューサーによって生成された各メッセージは、1 つのパーティションにのみ送信されます。 Kafka のパーティション番号は 0 から始まります。プロデューサーが 2 つのパーティション内のトピックにメッセージを送信する場合、このメッセージはパーティション 0 またはパーティション 1 のいずれかに格納されます。

では、特定のパーティションにメッセージを指定するにはどうすればよいでしょうか?

ここで、プロデューサーの送信ロジックを見てみましょう。その前に、ProducerRecord と呼ばれるものについて知っておく必要があります。これは何ですか？

ProducerRecord は、PR と呼ばれる基本データ情報をカプセル化してブローカーに送信されるキー/値のペアです。

内部構造

トピック（名前）
パーティションID (オプション)
キー（オプション）
価値

プロデューサー送信ロジック

1. パーティション ID が指定されている場合、PR は指定されたパーティションに送信されます。

2. パーティションIDが指定されていないがキーが指定されている場合、PRはハッシュ（キー）に従って対応するパーティションに送信されます。

3. パーティションIDもキーも指定されていない場合、PRはデフォルトのラウンドロビンモードを使用して各パーティションに送信します（コンシューマパーティションのデフォルトモードは範囲モードです）。

4. パーティション ID とキーの両方が指定されている場合、PR は指定されたパーティションにのみ送信されます (この時点ではキーは機能しません。コード ロジックによって決定されます)

注: パーティションには複数のコピーがありますが、このパーティションとプロデューサーおよびコンシューマー間のやり取りを担当する ReplicationLeader は 1 つだけです。

生産者からブローカーへの発送プロセス

Kafka クライアントがサーバーにデータを送信すると (一度に 1 つのメッセージではなく)、データはメモリ バッファーを通過します。 KafkaProducer を介して送信されたメッセージは、最初にクライアントのローカル キャッシュに入り、その後、メッセージがバッチに収集され、一度にブローカーに送信されます。この方法でのみパフォーマンスを向上させることができます。

プロデューサー向け共通設定

#kafka アドレス、つまりブローカー アドレス
ブートストラップサーバー
 
 #プロデューサーがリーダーにデータを送信するときに、request.required.acks パラメータを使用して、データの信頼性レベル (0、1、またはall ) を設定できます。
アク
 
 #リクエストが失敗した場合、プロデューサーは 0 回を指定して自動的に再試行します。再試行が有効になっている場合、メッセージが重複する可能性があります
再試行
 
 #各パーティションの未送信メッセージの合計バイトサイズ、単位: バイト。値が設定値を超えると、データがサーバーに送信されます。デフォルト値は16KBです
バッチサイズ  
 
 # デフォルト値は 0 で、 batch.sizeバッファ スペースがいっぱいでなくてもメッセージはすぐに送信されます。リクエスト数を減らしたい場合は、linger.ms を #0 より大きく設定します。つまり、メッセージがバッファ内に保持される時間です。設定値を超えるとサーバーに送信されます。
 # 簡単に言えば、ずっと前に送信されるはずだったメッセージは、少なくとも linger.ms 時間待機させられます。この間に蓄積されるメッセージが増えるため、バッチ送信によってリクエストが削減されます。
 #バッチがいっぱいになった場合、またはlinger.msが上限に達した場合、どちらかが満たされると送信されます
リンガー
 
 # buffer.memory は、Kafka Producer が使用できるメモリ バッファーのサイズを制限するために使用されます。デフォルト値は 32MB です。
 # buffer.memory の設定が小さすぎると、メッセージはメモリ バッファにすぐに書き込まれますが、Sender スレッドがメッセージを Kafka サーバーに送信する時間がない可能性があります。
 # メモリ バッファはすぐにいっぱいになり、いっぱいになるとユーザー スレッドがブロックされ、Kafka へのメッセージの書き込みができなくなります。
 # buffer.memory はbatch.sizeより大きくする必要があります。そうでない場合は、メモリ不足を示すエラー メッセージが表示されます。物理メモリを超えないように、実際の状況に応じて調整してください。
バッファメモリ
 
 #キーシリアライザーは、ユーザーが提供するキーと値のオブジェクト ProducerRecordをシリアル化します。 key.serializerは設定する必要があります。
 # メッセージにキーが指定されていません。シリアライザーは、org.apache.kafka.common.serialization.Serializer インターフェースを実装し、#キーをバイト配列にシリアル化するクラスである必要があります。
キー.シリアライザー
値シリアライザー

Kafka の Consumer メカニズムとパーティション戦略の説明

消費者はどのようなモードでブローカーからデータを取得しますか?ブローカーが積極的にプッシュするのではなく、プル モードになっているのはなぜですか?

答えは記事の冒頭の写真でご覧いただけます。コンシューマーは Pull メソッドを使用してブローカーのパーティションからデータを取得します。なぜプッシュモードではなくプルモードなのでしょうか?プルモードは、消費者の消費能力に応じて調整できます。消費者によってパフォーマンスは異なります。ブローカーにデータがない場合、コンシューマーはタイムアウトを設定してブロックし、戻る前にしばらく待機することができます。ただし、ブローカーが積極的にプッシュする場合、プッシュの利点はメッセージを迅速に処理できることですが、コンシューマーが処理できず、メッセージが蓄積され、遅延が発生する可能性が高くなります。

消費者はどのパーティションから消費しますか?

トピックには複数のパーティションがあり、コンシューマー グループには複数のコンシューマーが存在することがわかります。どのように割り当てられるのでしょうか?トピックには複数のパーティション (リーダー パーティション) があるため、複数のコンシューマーが存在する場合があります。パーティション リーダーは、コンシューマー グループ内のコンシューマーによって消費される可能性があります。

では、消費者はどのパーティションから消費するのでしょうか?

戦略 1: ラウンドロビン (RoundRobinAssignor はデフォルトの戦略ではありません)。ラウンドロビン割り当ては、消費者グループに従って実行されます。同じ消費者グループが、さまざまなトピックを同じ方法で監視します。すべてのパーティションとすべてのコンシューマーがリストされます。したがって、コンシューマー グループ内のサブスクライブされたトピックは同じである必要があります。トピックが異なると、割り当てが不均等になります。たとえば、次の例をご覧ください。

 #同じグループには7つのパーティションと2つのコンシューマーがあります
topic-p0/topic-p1/topic-p2/topic-p3/topic-p4/topic-p5/topic-p6 (パーティション) 
 
 c-1: トピックp0/トピックp2/トピックp4/トピックp6 (コンシューマー1) 
 
 c-2:トピック-p1/トピック-p3/トピック-p5 (コンシューマー2)

これの欠点は何ですか?同じコンシューマー グループ内でサブスクライブされたメッセージが異なる場合、パーティションを実行するときに割り当てがラウンドロビンで行われず、パーティションの割り当てが不均一になる可能性があります。たとえば、3 つのコンシューマー C0、C1、C2 があり、合計 3 つのトピック t0、t1、t2 をサブスクライブしているとします。このとき、t0 には 1 つのパーティション (p0)、t1 には 2 つのパーティション (p0、p1)、t2 には 3 つのパーティション (p0、p1、p2) があります。コンシューマー C0 はトピック t0 をサブスクライブし、コンシューマー C1 はトピック t0 と t1 をサブスクライブし、コンシューマー C2 は t0、t1、および t2 をサブスクライブします。これはポーリング メカニズムであるため、C0 が T0 をサブスクライブする場合、C1 は T0 をサブスクライブできませんが、T1 をサブスクライブできます。 C2 も T0 をサブスクライブすることはできませんが、T1 と T2 は両方とも T0 をサブスクライブできます。この時点では、C2 のみが T2 をサブスクライブしており、他の C0 と C1 は表示されません。このとき、T2 のメッセージはコンシューマーである C2 によって消費されます。この状況は不均等分配の問題です。

戦略 2、範囲 (RangeAssignor のデフォルト戦略)、トピックごとに割り当てます。均等に分散されていない場合は、最初のコンシューマーにさらに多くのパーティションが割り当てられます。異なるトピックを聞いている消費者には影響しません。この戦略の欠点は何ですか?トピックが 1 つだけの場合、c-1 がさらに 1 つのパーティションを消費しても、大きな影響はありません。トピックが複数ある場合、各トピックに対して、コンシューマー C-1 は 1 つのパーティションをさらに消費します。トピックの数が増えると、消費されるパーティションの数が増え、パフォーマンスが低下します。

[インタビュー質問] 消費者再分配戦略とオフセット維持メカニズム

リバランス操作とは何ですか?

Kafka はどのようにしてトピックの下にあるすべてのパーティションを各コンシューマーに均等に分散し、メッセージの消費速度を可能な限り速くするのでしょうか?これがバランスです。再バランス調整は、実際にはパーティションを再配分して、パーティションの配分が再びバランスの取れた状態になるようにすることです。下の図に示すように、A と B という 2 つのコンシューマーがあります。3 番目のメンバー C が参加すると、Kafka はリバランスをトリガーします。再分配戦略は、A、B、C を再パーティション化することです。再バランス後の分配は依然として公平であり、各 Consumer インスタンスは 2 つのパーティションの使用権を取得します。

消費者が消費中に突然クラッシュした場合、回復後に消費はどこから始まるのでしょうか?どのような問題が発生するでしょうか?

コンシューマーはオフセットを記録し、障害が回復した後、ここから消費を続けます。では、オフセットはどこに記録されるのでしょうか? Zookeeper とローカルに記録されます。新しいバージョンでは、オフセットがデフォルトで _consumer_offsets という名前の Kafka の組み込みトピックに含まれるようになります。デフォルトでは、このトピックには 50 個のパーティションがあり、各パーティションには 3 つのレプリカがあります。パーティションの数は、パラメータ offset.topic.num.partition によって設定されます。 groupid のハッシュ値とこのパラメータの係数は、コンシューマ グループによって消費されるオフセットが保存される _consumer_offsets トピックのパーティションを決定するために使用されます。これは、コンシューマー グループ名 + トピック + パーティションによって決定され、一意のオフセット キーが決定され、対応する値が取得されます。