在宅勤務中に整理された Kafka の知識ポイントの完全なリスト

Kakfa は、BAT、ByteDance、Meituan、Netflix、Airbnb、Twitter など、国内外の大手企業で広く使用されています。本日は、この記事を通じて Kafka の仕組みを詳しく見ていきます。

[[315253]]

画像はPexelsより

Kafka の概要

Kakfa は、パブリッシュ/サブスクライブ モードに基づく分散メッセージ キューであり、主にビッグ データのリアルタイム処理の分野で使用されます。

メッセージキュー

従来のメッセージ キューと新しいメッセージ キュー モードは次のとおりです。

上記は従来のメッセージ キューです。たとえば、ユーザーが情報を登録する場合、ユーザー情報がデータベースに書き込まれた後、その背後でテキストメッセージの送信などの他のプロセスが実行されます。これらのプロセスが完了するまで待ってから、ユーザーに戻る必要があります。

新しいキューを使用すると、たとえばユーザー登録情報などのデータが直接データベースにスローされ、成功がユーザーに直接返されます。

メッセージ キューを使用する利点は次のとおりです。

• デカップリング
• 回復可能性
• バッファ
• 柔軟性とピーク時の処理能力
• 非同期通信

メッセージ キュー モデルは次のとおりです。

① ポイントツーポイントモード：メッセージプロデューサーがメッセージをメッセージキューに送信し、メッセージコンシューマーがメッセージをキューから取り出して消費します。メッセージは消費されると、キューに保存されなくなります。

したがって、メッセージ コンシューマーがすでに消費されたメッセージを消費することは不可能です。キューは複数のコンシューマーをサポートしますが、メッセージについては 1 つのコンシューマーのみがそれを消費できます。複数の消費者に送信する場合は、メッセージを複数回送信する必要があります。

② パブリッシュ/サブスクライブモード（1対多、コンシューマーはデータを消費した後メッセージをクリアしません）：メッセージプロデューサーがトピックにメッセージをパブリッシュし、複数のメッセージコンシューマー（サブスクライブ）が同時にメッセージを消費します。

ポイントツーポイント方式とは異なり、トピックに公開されたメッセージはすべてのサブスクライバーによって消費されます。ただし、ストレージ システムではないため、データの保持期間はデフォルトで 7 日間に制限されています。

Kafka はこのモデルに基づいています。方法は2つあります。 1 つは、プロデューサーがコンシューマーにメッセージをプッシュするのではなく、コンシューマーがメッセージをアクティブに消費 (プル) することです。もう 1 つは、公式アカウントと同様に、プロデューサーが積極的に消費者にメッセージをプッシュすることです。

Kafka インフラストラクチャ

Kafka のアーキテクチャは次のとおりです。

Kafka のインフラストラクチャは主にブローカー、プロデューサー、コンシューマー グループで構成されており、現在は ZooKeeper も含まれます。

プロデューサーはメッセージの送信を担当し、ブローカーはメッセージのバッファリングを担当し、トピックはブローカーで作成でき、各トピックにはパーティションとレプリケーションの概念があります。

コンシューマー グループはメッセージの処理を担当します。同じコンシューマー グループ内のコンシューマーは、同じパーティション内のデータを消費できません。

消費者グループは主に消費能力の向上に使用されます。たとえば、以前は 1 人の消費者が 100 個のデータを消費していましたが、現在は 2 人の消費者が同じ 100 個のデータを消費しており、消費能力が向上します。

したがって、コンシューマー グループ内のコンシューマーの数はパーティションの数よりも少なくする必要があります。そうでないと、一部のコンシューマーには消費するパーティションがなくなり、リソースが無駄になります。

注: 異なるコンシューマー グループのコンシューマーが同じパーティション データを消費できます。

Kakfa がコンポーネント クラスターを形成する場合は、ZooKeeper に登録するだけで済みます。 ZooKeeper は、メッセージ消費の進行状況、オフセット、または消費位置も保持します。

• 0.9 より前のバージョンでは、オフセットは ZooKeeper に保存されていました。
• 0.9 以降のバージョン オフセットは Kafka に保存されます。 Kafka は、オフセット データを格納するために特別に使用されるシステム トピックを定義します。

なぜ変更するのですか?主な理由は、オフセットの頻繁な変更が ZooKeeper に大きな負担をかけ、Kafka 自体の処理も複雑になるためです。

Kafkaをインストールする

①Kafkaをインストールするには、インストールパッケージを解凍するだけでインストールが完了します。

tar -zxvf kafka_2.11-2.1.1.tgz -C /usr/local/

②設定ファイルを表示します。

 [root@es1 config]# pwd
 /usr/ローカル/kafka/config
 [root@es1 config]# ll
合計 84
 -rw-r --r--。 1 ルート ルート 906 2019年2月8日 connect-console-sink.properties  
 -rw-r --r--。 1 ルート ルート 909 2019年2月8日 connect-console-source.properties  
 -rw-r --r--。 1 ルート ルート 5321 2019年2月8日 connect-distributed.properties  
 -rw-r --r--。 1 ルート ルート 883 2019年2月8日 connect-file-sink.properties  
 -rw-r --r--。 1 ルート ルート 881 2019年2月8日 connect-file-source.properties  
 -rw-r --r--。 1 ルート ルート 1111 2019年2月8日 connect-log4j.properties  
 -rw-r --r--。 1 ルート ルート 2262 2019年2月8日 connect-standalone.properties  
 -rw-r --r--。 1 ルート ルート 1221 2019年2月8日 consumer.properties  
 -rw-r --r--。 1 ルート ルート 4727 2019年2月8日 log4j.properties  
 -rw-r --r--。 1 ルート ルート 1925 2月 8 2019 producer.properties  
 -rw-r --r--。 1 ルート ルート 6865 1月16日 22:00 server-1.properties  
 -rw-r --r--。 1 ルート ルート 6865 1月16日 22:00 server-2.properties  
 -rw-r --r--。 1 ルート ルート 6873 1月16日 03:57 server.properties  
 -rw-r --r--。 1 ルート ルート 1032 2019年2月8日 tools-log4j.properties  
 -rw-r --r--。 1 ルート ルート 1169 2019年2月8日 trogdor.conf  
 -rw-r --r--。 1 ルート ルート 1023 2019年2月8日 zookeeper.properties  

③設定ファイルserver.propertiesを変更します。

Kafka クラスター内の各ノードの一意の識別子である broker.id を設定します。

④Kafkaのデータ保存パスを設定します。

注意: このディレクトリの下に Kafka 以外のディレクトリが存在してはいけません。そうしないと、Kafka クラスターの起動に失敗します。

⑤トピックを削除できるかどうかを設定します。デフォルトでは、Kafka のトピックは削除できません。

⑥Kafkaデータのデフォルトの保存期間は7日間です。

⑦ログファイルの最大サイズ。ログファイルが1Gを超えると、新しいファイルが作成されます。

⑧Kafka が接続する ZooKeeper アドレスと、Kafka への接続のタイムアウト。

⑨デフォルトのパーティション数。

Kafkaを起動する

①起動方法1：Kafkaは単一のノードでのみ起動できるため、各Kakfaノードを手動で起動する必要があります。以下の方法は、ブロッキング方式で起動する方法です。

②起動方法2：ガーディアン起動、推奨。

Kafka オペレーション

① 現在の Kafka クラスター内の既存のトピックを表示します。

注: これは Kafka ではなく ZooKeeper に接続します。

②トピックを作成し、シャードとレプリカの数を指定します。

注: replication-factor はレプリカの数、replication-factor はパーティションの数、topic はトピック名です。

現在の Kafka クラスターにブローカー ノードが 3 つしかない場合、最大レプリケ​​ーション係数は 3 です。次の例では 4 つのレプリカが作成され、エラーが発生します。

③トピックを削除します。

④トピック情報を閲覧します。

プロデューサーを起動してメッセージを生成する

Kafka にはプロデューサー クライアントとコンシューマー クライアントが付属しています。

① プロデューサーを起動し、ポート9092で接続されたKafkaクラスターに注目します。

② コンシューマーを起動します。接続は引き続きポート 9092 に行われることに注意してください。バージョン 0.9 より前は、接続はポート 2181 に行われました。

ここで、テストするために 2 つのコンシューマーを起動します。

注: コンシューマー グループ構成ファイルを指定しない場合、各コンシューマーはデフォルトで異なるコンシューマー グループに属します。

③メッセージを送信すると、すべての消費者がメッセージを受信できることがわかります。

④Kakfaの実際のデータ。

Kafka アーキテクチャの詳細

Kafka はメッセージのグローバルな順序を保証することはできませんが、コンシューマーは異なるパーティションでメッセージをランダムに消費するため、パーティション内のメッセージの順序のみを保証できます。

Kafkaワークフロー

Kafka 内のメッセージはトピックによって分類されます。プロデューサーはメッセージを生成し、コンシューマーはトピックに基づいてメッセージを消費します。

トピックは論理的な概念ですが、パーティションは物理的な概念です。各パーティションにはレプリカの概念があります。

各パーティションは、プロデューサーによって生成されたデータを保存するログ ファイルに対応します。プロデューサーによって生成されたデータは、ログ ファイルの末尾に継続的に追加されます。

各データには独自のオフセットがあり、コンシューマーは消費したオフセットをリアルタイムで記録し、エラーが発生した場合に最後の位置から消費を続行できるようにします。このオフセットはインデックス ファイルに保存されます。

Kafka のオフセットはパーティション内では順序付けられますが、異なるパーティション間では順序付けられません。 Kafka はデータのグローバルな順序を保証しません。

カフカの原則

プロデューサーによって生成されたメッセージはログ ファイルの末尾に継続的に追加されるため、ログ ファイルが大きくなりすぎてデータの配置が非効率になるのを防ぐために、Kafka はシャーディングとインデックス作成のメカニズムを使用して各パーティションを複数のセグメントに分割します。各セグメントは、インデックス ファイルとログ ファイルの 2 つのファイルに対応します。

2 つのファイルは同じフォルダーにあります。フォルダーの命名規則は、トピック名 + パーティション番号です。

インデックス ファイルとログ ファイルのファイル名は、現在のインデックス内の最小データのオフセットです。 Kafka はどのようにしてデータを素早く消費するのでしょうか?

インデックス ファイルに保存されているデータのインデックス情報。最初の列はオフセットで、2 番目の列はデータに対応するログ ファイル内のオフセットです。ファイルを読み取り、seek() を使用して現在のマウスの位置を設定する場合と同様に、データをより速く見つけることができます。

オフセットが 3 のデータを使用する場合は、まずバイナリ検索を使用してデータが含まれているインデックス ファイルを見つけ、次にインデックス内のオフセットを使用してログ ファイル内のデータのオフセットを見つけます。こうすることで、データをすばやく見つけて利用できるようになります。

したがって、Kakfa はデータをディスクに保存しますが、読み取り速度は依然として非常に高速です。

Kafka プロデューサーとコンシューマー

カフカプロデューサー

Kafka のパーティションの役割: Kafka パーティションの主な目的は、読み取りと書き込みがパーティション単位で行われるため、同時実行性を実現し、パフォーマンスを向上させることです。

では、プロデューサーはどのパーティションにメッセージを送信するのでしょうか?

クライアントでパーティションを指定します。

ポーリング (推奨) メッセージ 1 は p1 に、メッセージ 2 は p2 に、メッセージ 3 は p3 に、メッセージ 4 は p1 に、メッセージ 5 は p2 に、メッセージ 6 は p3 に送られます...

Kafka はどのようにしてデータの信頼性を確保するのでしょうか?

Kafka はどのようにしてデータの信頼性を確保するのでしょうか?スルーアク！

プロデューサーによって送信されたデータが指定されたトピックに確実に送信されるようにするには、トピックの各パーティションが、プロデューサーによって送信されたデータを受信した後、プロデューサーに Ack (受信確認) を送信する必要があります。プロデューサーが Ack を受信した場合は、次の送信ラウンドに進みます。それ以外の場合は、データを再送信します。

では、Kafka はいつプロデューサーに Ack を送信するのでしょうか?フォロワーとリーダーが同期されていることを確認するために、リーダーはプロデューサーに Ack を送信します。これにより、リーダーが電話を切った後、フォロワーから新しいリーダーが選出され、データが失われることがなくなります。

同期が完了した後、何人のフォロワーが Ack を送信しますか?

• 解決策 1: パケットの半分の同期が完了すると、Ack が送信されます。
• 解決策 2: すべての同期が完了した後にのみ Ack が送信されます (Kafka はこの方法を使用します)。

2 番目のソリューションを採用した後、次のシナリオを想像してください。リーダーがデータを受信し、すべてのフォロワーがデータの同期を開始しますが、1 つのフォロワーが何らかの障害のために同期を完了できません。その後、リーダーは Ack を送信する前に同期が完了するまで待機する必要があります。

これは効率に大きく影響します。この問題を解決するにはどうすればいいでしょうか?

リーダーは動的な ISR リスト (同期レプリカの役割) を維持し、このリスト内のフォロワーのみをリーダーと同期する必要があります。

ISR 内のフォロワーがデータ同期を完了すると、リーダーはプロデューサーに Ack を送信します。フォロワーがリーダーと長時間データを同期できない場合、フォロワーは ISR から削除されます。この時間しきい値もカスタマイズ可能です。

同様に、リーダーが失敗した場合、ISR から新しいリーダーが選出されます。

ISR ノードを選択するにはどうすればよいでしょうか?まず第一に、コミュニケーションの時間が速く、リーダーとのコミュニケーションが素早く完了する必要があります。デフォルトの時間は 10 秒です。

次に、リーダーのデータギャップを確認します。デフォルトのメッセージ数は 10,000 です (以降のバージョンでは削除されました)。

なぜ削除するのですか? Kafka はメッセージをバッチで送信するため、リーダーはメッセージを即座に受け入れますが、フォロワーはまだメッセージをプルしていないため、頻繁に ISR から抜け出して参加します。このデータは ZooKeeper とメモリに保存されるため、ZooKeeper とメモリは頻繁に更新されます。

ただし、重要度の低いデータの場合、データの信頼性要件はそれほど高くなく、少量のデータ損失は許容されるため、ISR 内のすべてのフォロワーが正常に受け入れるまで待つ必要はありません。

したがって、Kafka はユーザーに 3 つのレベルの信頼性を提供し、ユーザーは信頼性とレイテンシに基づいてトレードオフを行うことができます。この設定は、Kafka: Ack パラメータ設定の生成時に設定されます。

①Acksが0の場合：プロデューサーはAckを待たずにトピックにデータを投げ込むだけです。データが失われる可能性が非常に高くなります。

②Ackが1の場合：リーダーはディスクに書き込まれた後にAckを返し、データは失われます。同期が完了した後にリーダーに障害が発生すると、データは失われます。

③Ackが-1（すべて）：リーダーとフォロワー（ISR）がディスクに書き込んだ後にのみAckが返されるため、データの重複が発生する可能性があります。リーダーが書き込みを完了し、フォロワーが同期を完了したにもかかわらず、Ack を返すときに障害が発生した場合、データの重複が発生する可能性があります。

極端な場合には、フォロワーとリーダー間の通信が非常に遅い場合、ISR にリーダー ノードが 1 つしかないなど、データ損失が発生する可能性もあります。

このとき、リーダーはディスク プッシュを完了した後に Ack を返します。この時点でリーダーに障害が発生すると、データが失われます。

Kafka は消費データの一貫性をどのように確保しますか?

Kafka は消費データの一貫性をどのように確保しますか? HW経由:

• LEO: 各フォロワーの最大オフセットを指します。
• HW (ハイ ウォーター マーク): コンシューマーが確認できる最大オフセット、つまり LSR キュー内の最小の LEO を指します。つまり、コンシューマーは 1 から 6 までのデータのみを確認でき、それ以降のデータは確認も使用もできません。

リーダーが失敗する状況を回避するために、たとえば、現在のコンシューマーがデータ 8 を消費した後、リーダーが失敗します。このとき、たとえば f2 がリーダーになった場合、f2 にはデータ 9 がまったくないため、コンシューマーはエラーを報告します。したがって、HW パラメータは、上記の問題を回避するために、最小限のデータのみを消費者に公開するように設計されています。

HW はデータ ストレージの一貫性を保証します。

①フォロワーの障害：フォロワーが障害を起こすと、一時的にLSRから追い出されます。フォロワーが回復すると、ローカル ディスクに記録された最後の HW が読み取られ、ログ ファイルの HW より高い部分が切り取られます。 HW からのリーダーとの同期を開始します。フォロワーの LEO がパーティションの HW 以上の場合、つまりフォロワーがリーダーに追いつくと、LSR に再参加できます。

② リーダーの故障：リーダーが故障した場合、ISRから新しいリーダーが選出されます。その後、複数のレプリカ間でデータの一貫性を確保するために、残りのフォロワーはまずそれぞれのログ ファイルの HW より高い部分を切り捨て (新しいリーダーはそれを切り捨てません)、次に新しいリーダーからのデータを同期します。

注: これは、複数のコピー間でのデータ保存の一貫性を確保するためのものであり、データが失われたり重複したりしないことを保証するものではありません。

データが重複しないようにするために、正確に 1 回 (べき等性) を実行します。

• Ack が -1 に設定されている場合、データ損失は保証されますが、データの重複が発生する可能性があります (少なくとも 1 回)。
• Ack を 0 に設定すると、データが重複しないことは保証されますが、データが失われないことは保証されません (最大 1 回)。

しかし、もしケーキを食べてケーキも残したいとしたらどうでしょうか?ここで Exact Once が登場します。

バージョン 0.11 以降では、Kakfa クラスター内のデータ重複を解決するためにべき等性が導入されました。バージョン 0.11 より前では、コンシューマーが独自に処理していました。

べき等性が有効な場合、Ack のデフォルト値は -1 になり、Kafka は各メッセージに Seqnumber を割り当てる代わりに、各プロデューサーに Pid を割り当てます。

Pid、Partition、Seqnumber が同じ場合、Kafka はそれを重複データと見なし、ディスクに保存しません。

ただし、プロデューサーがクラッシュすると、データの重複が発生する可能性があります。したがって、べき等性は単一セッションの単一パーティション内のデータ重複を解決しますが、パーティション間またはセッション間のデータ重複は解決できません。

Kafka コンシューマー

①摂取方法

メッセージ キュー内のメッセージを使用するには、プッシュ (WeChat 公式アカウント) とプル (kafka) の 2 つの方法があります。

プッシュ モードは、消費送信レートがブローカーによって決定され、その目標がメッセージをできるだけ早く配信することであるため、消費レートの異なるコンシューマーに適応するのが困難です。

ただし、これにより、消費者がメッセージを処理する時間がなくなる可能性が高まり、典型的な兆候としてサービス拒否やネットワーク輻輳が発生します。 Pull メソッドは、コンシューマーの消費能力に基づいて適切な速度でメッセージを消費できます。

Pull モードの欠点は、Kafka にデータがない場合、コンシューマーが無限ループに陥り、空のデータを返し続ける可能性があることです。この問題に対処するために、Kafka コンシューマーはデータを消費するときに Timeout パラメータを渡します。その時点で消費するデータがない場合、コンシューマーはしばらく待ってから戻ります。

②パーティション割り当て戦略

コンシューマー グループには複数のコンシューマーが含まれ、トピックには複数のパーティションが含まれます。したがって、パーティションの割り当ての問題、つまりどのパーティションがどのコンシューマーによって消費されるかを決定する問題が必ず関係してきます。

Kafka には 2 つの方法があります。1 つはトピック グループに有効なラウンドロビン、もう 1 つは単一のトピックに有効な範囲です。

ポーリング: 前提条件として、コンシューマー内のすべてのコンシューマーが同じトピックをサブスクライブする必要があります。そうしないと問題が発生します。デフォルト以外の方法。

同じコンシューマー グループ内のコンシューマーは、同時に同じパーティションを消費することはできません。たとえば、3 人のコンシューマーがトピックの 9 つのパーティションを消費します。

コンシューマー グループに 2 つのコンシューマーが存在する場合、このコンシューマー グループは同時に 2 つのトピックを消費し、各トピックには 3 つのパーティションがあります。

まず、2 つのトピックが 1 つのトピックとして扱われ、次にトピックとパーティションに従ってハッシュされ、最後にハッシュに従ってソートされます。次に、ポーリングはコンシューマー グループ内の 2 つのコンシューマーに割り当てられます。

サブスクリプションが次の方法で行われた場合はどうなりますか?たとえば、トピックが 3 つあり、各トピックに 3 つのパーティションがあり、コンシューマー グループにコンシューマーが 2 つあるとします。

コンシューマー 1 はトピック 1 とトピック 2 をサブスクライブし、コンシューマー 2 はトピック 2 とトピック 3 をサブスクライブします。このようなシナリオでは、ポーリングを使用してトピックをサブスクライブする際に問題が発生します。

サブスクリプションが次の方法で行われた場合はどうなりますか?たとえば、トピックが 2 つあり、各トピックに 3 つのパーティションがあり、コンシューマー グループに 2 つのコンシューマーがあり、コンシューマー 1 がトピック 1 をサブスクライブし、コンシューマー 2 がトピック 2 をサブスクライブしているとします。ポーリングを使用してトピックをサブスクライブする場合にも問題が発生します。

したがって、私たちは常に、トピックをサブスクライブするためにポーリングを使用する前提は、コンシューマー グループ内のすべてのコンシューマーが同じトピックをサブスクライブすることであると強調してきました。したがって、ポーリング方式は Kafka のデフォルトの方式ではありません。範囲は、デフォルトの割り当て方法である単一のトピックに従って分割されます。

Range の問題は、消費者データが不均衡であることです。たとえば、次の例では、コンシューマー グループが 2 つのトピックをサブスクライブし、コンシューマー 1 は 4 つのパーティションを消費しますが、他のコンシューマーは 2 つのパーティションのみを消費します。

パーティション戦略はいつ発動されますか?コンシューマー グループ内のコンシューマーの数が変更されると、コンシューマー グループ内のコンシューマーの追加や削減などのパーティション戦略の調整がトリガーされます。

③オフセットを維持する

消費者は消費の過程で停電やその他の障害を経験する可能性があるため、復旧後は障害前の位置から消費を継続する必要があります。したがって、消費者は、障害が回復した後も消費を継続できるように、消費したオフセットを記録する必要があります。

Offset が保存される場所は 2 つあり、1 つは ZooKeeper、もう 1 つは Kafka です。まず、ZooKeeper に保存されているオフセットを見てみましょう。一意のオフセットは、コンシューマー グループ、トピック、パーティションの 3 つの要素によって決定されます。

したがって、コンシューマー グループ内のコンシューマーが失敗した場合でも、コンシューマーはオフセットを取得できます。

コントローラー ノードは ZooKeeper と通信してデータを同期します。最初に起動したノードがコントローラーに登録され、コントローラーになります。他のノードとコントローラー情報は同期されたままになります。

④消費者団体の事例

コンシューマー グループ ID を変更します。

コンシューマーを起動して 3 つのデータを送信する:

コンシューマーを起動するコンシューマー グループを指定します。 3 つのコンシューマーを起動すると、各コンシューマーが 1 つのデータを消費することがわかります。

デモでは、異なるグループが同じトピックを消費でき、両方のグループのコンシューマーが同じデータを消費していることがわかります。もう一度コンシューマーを開始します。このコンシューマーは別のコンシューマー グループに属します。

Kafka の効率的な読み取りと書き込みのメカニズム

分散展開

マルチノード並列操作。

ディスクへの順次書き込み

Kafka のプロデューサーによって生成されたデータはログ ファイルに書き込まれ、書き込みプロセス中にファイルの末尾に追加されます。これは公式サイトのデータに示されているように、順次書き込みです。

同じディスクの場合、シーケンシャル書き込み速度は 600M/S に達しますが、ランダム書き込み速度はわずか 100K/S です。これはディスクの機械的な構造に関係しています。シーケンシャル書き込みが高速な理由は、ヘッドのアドレス指定にかかる時間が大幅に節約されるためです。

ゼロコピー技術

通常の状況では、データは最初にカーネル空間に読み込まれ、次にカーネル空間からユーザー空間に読み込まれ、その後、オペレーティング システムの IO インターフェイスが呼び出されてデータがカーネル空間に書き込まれ、最後にハード ディスクに書き込まれます。

Kafka は IO ストリームをカーネル空間で直接転送することでこれを実現するため、Kafka のパフォーマンスは非常に高くなります。

Kafka における ZooKeeper の役割

Kafka クラスターでは、ブローカーがコントローラーとして選出され、クラスター ブローカーのオンラインおよびオフライン操作、すべてのトピックのパーティション レプリカの割り当て、およびリーダー選出の管理を担当します。