3分でKafkaを完全に理解する

[[406253]]

1. Kafkaを理解する

Kafka とは何でしょうか?それは何に使われますか?

公式の定義は次のとおりです。

Kafka は、リアルタイム データ パイプラインとストリーミング アプリの構築に使用されます。水平方向にスケーラブルで、フォールト トレラント、非常に高速であり、何千もの企業で実稼働されています。

直訳すると、水平展開可能で信頼性の高いリアルタイムデータ処理システムということです！

リアルタイムデータ処理は名前から簡単に理解できます。データをリアルタイムで処理することを意味します。現在普及しているマイクロサービス開発において、最も一般的に使用されているリアルタイム データ処理プラットフォームは、RabbitMQ や RocketMQ などのメッセージ ミドルウェアです。

これらのミドルウェアには主に 2 つの機能があります。

• サービスの分離
• 交通ピークカット

Web アプリケーション開発の初期には、リクエスト数が急増したときに、処理するデータをキュー チャネルにプッシュし、別のスレッドを開始してキュー内のデータを継続的にローテーションおよびプルすることで、プログラムの実行効率を高めていました。

ただし、リクエストの数が増加し続け、キュー チャネル内のデータが常に高負荷状態になると、アプリケーションのメモリ使用量が非常に高くなります。注意しないと、メモリが不足し、プログラム メモリのオーバーフローが発生し、サービスが利用できなくなります。

ビジネス量が拡大し続けると、このモデルではアプリケーション内のニーズを満たすことができなくなります。そのため、ActiveMQ、RabbitMQ、RocketMQなどのさまざまなメッセージミドルウェアが誕生しました。

このモデルの本質は、プッシュされるデータが現在のアプリケーションのメモリに保存されるのではなく、データ処理を専門に担当する別のアプリケーションに保存され、それによってサービス分離が実現されることです。

メッセージ ミドルウェア: 主な役割は、メッセージを受信して​​ディスクに保存できるようにすることです。他のサービスに障害が発生しても、データは失われません。データ消費量を監視することもできます。

アプリケーション: メッセージをメッセージ ミドルウェアにプッシュし、スレッドがメッセージ ミドルウェアから継続的にデータをプルして消費を確認できるようにします。

メッセージ ミドルウェアの導入後は、サービス開発全体が簡素化され、全員が自分のタスクに責任を持つようになります。

Kafka は本質的にはメッセージ ミドルウェアの一種です。 Kafka は LinkedIn から生まれ、2010 年に GitHub にオープンソース化されました。

データ パイプラインの問題を解決するために、LinkedIn の開発チームは 2010 年頃にデータ交換に ActiveMQ を最初に採用しました。当時、ActiveMQ は LinkedIn のデータ転送システムの要件を満たすにはほど遠いものでした。さまざまな欠陥により、メッセージがブロックされたり、サービスが利用できなくなったりすることがよくありました。この問題を解決するために、LinkedIn は独自のメッセージング システムを開発することを決定し、Kafka が誕生しました。

LinkedIn では、Kafka は 1 日あたり数十億件のメッセージのメトリックとユーザー アクティビティの追跡を効果的に処理できます。その強力な処理能力は業界で認められ、ビッグデータ パイプラインの推奨テクノロジーとなっています。

2. アーキテクチャの紹介

まずは写真を見てみましょう。下の図は、Kafka の生成と消費のコア アーキテクチャ モデルです。

これらの概念を理解していなくても問題ありません。一緒に説明します。

• プロデューサー: プロデューサーはプロデューサーであり、メッセージのプロデューサーであり、メッセージのエントリポイントです。
• ブローカー: ブローカーは Kafka インスタンスです。各サーバーには 1 つ以上の Kafka インスタンスがあります。簡単に言えば、Kafka サーバーです。 Kafka クラスターはクラスターを意味します。
• トピック: メッセージの件名。メッセージ キューとして理解できます。 Kafka データはトピックに保存されます。各ブローカーで複数のトピックを作成できます。
• パーティション: トピック パーティション。各トピックには複数のパーティションを含めることができます。パーティションの役割は、ロードを実行し、Kafka のスループットを向上させることです。異なるパーティション内の同じトピックのデータは重複せず、パーティションは 1 つずつフォルダーで表されます。
• レプリケーション: 各パーティションには複数のレプリカがあります。レプリカの役割はバックアップとして機能することです。プライマリ パーティション (リーダー) は、スレーブ パーティション (フォロワー) にデータを同期します。プライマリ パーティション (リーダー) に障害が発生すると、バックアップ パーティション (フォロワー) が選択され、引き継いでリーダーになります。 Kafka のレプリカのデフォルトの最大数は 10 であり、レプリカの数はブローカーの数より大きくすることはできません。フォロワーとリーダーは必ず異なるマシン上に存在し、同じマシンには同じパーティションのレプリカを 1 つだけ保存できます。
• メッセージ: 送信された各メッセージの本文。
• コンシューマー: コンシューマーはメッセージの消費者であり、メッセージの出口です。
• コンシューマー グループ: 複数のコンシューマー グループを 1 つのコンシューマー グループに結合できます。 Kafka の設計では、同じパーティション内のデータは、コンシューマー グループ内の 1 つのコンシューマーのみが使用できます。同じコンシューマー グループ内のコンシューマーは、同じトピックの異なるパーティションからデータを消費できるため、Kafka のスループットも向上します。
• Zookeeper: Kafka クラスターは、システムの可用性を確保するために、クラスターのメタデータを保存するために Zookeeper に依存しています。

つまり、Kafka は本質的にメッセージング システムです。ほとんどのメッセージング システムと同様に、その主な機能は次のとおりです。

• プッシュプルモデルを使用して生産者と消費者を分離する
• メッセージングシステム内のメッセージデータの永続性を提供し、複数の消費者が
• 高可用性クラスタサービス、マスタースレーブモードを提供し、水平拡張をサポートします。

ActiveMQ、RabbitMQ、RocketMQ との違いは、**パーティション** の概念があることです。

このパーティションは、作成するトピックに 5 つのパーティションがある場合、一度に 1,000 個のデータを Kafka にプッシュすると、これらの 1,000 個のデータはデフォルトで 5 つのパーティションに分散され、各パーティションに 200 個のデータが格納されることを意味します。

これを行う目的は、消費者がさまざまなパーティションからデータを簡単に取得できるようにすることです。同時に 5 つのスレッドを開始してデータをプルすると、各スレッドが 1 つのパーティションをプルするため、消費速度が非常に速くなります。

これが Kafka と他のメッセージング システムの最大の違いです。

2.1 データの送信

他のミドルウェアと同様に、Kafka は毎回リーダー パーティションにデータを送信し、それをディスクに順番に書き込みます。次に、リーダー パーティションはデータを各フォロワー パーティションに同期します。マスター パーティションに障害が発生しても、サービスの正常な動作には影響しません。

では、Kafka はどのようにして対応するパーティションにデータを書き込むのでしょうか? Kafka には次の原則があります。

1. データを書き込む際に、書き込むパーティションを指定できます。指定されている場合は、対応するパーティションに書き込みます

2. パーティションが指定されていないが、データのキーが設定されている場合は、キー値に従ってパーティションがハッシュ化されます。

3. パーティションが指定されておらず、キーも設定されていない場合は、ポーリングによってパーティションが選択されます。

2.2 消費データ

プロデューサーと同様に、コンシューマーが Kafka クラスターからメッセージをアクティブにプルすると、リーダー パーティションからデータもプルされます。

ここでは、「消費者グループ」という用語に注目する必要があります。

複数のコンシューマーのシナリオを考慮して、Kafka は複数のコンシューマーを持つコンシューマー グループを形成するように設計できます。同じコンシューマー グループ内のコンシューマーは、同じトピックの下にある異なるパーティションからのデータを消費できます。重複した消費を防ぐために、同じパーティションはコンシューマー グループ内のコンシューマーによってのみ消費されます。

ただし、異なるグループが同じパーティションからデータを消費する可能性があります。

これを次のように理解することができます。コンシューマー グループはクライアントであり、メッセージの消費を高速化するために、クライアントは多くのコンシューマーで構成できます。

ただし、グループ内のコンシューマーの数がパーティションの数より多い場合、多くのコンシューマーがアイドル状態になります。

パーティションの数がグループ内のコンシューマーの数より多い場合、1 人のコンシューマーが複数のパーティションの消費を担当することになり、消費パフォーマンスが不均一になります。

したがって、実際のアプリケーションでは、コンシューマー グループ内のコンシューマーの数をパーティションの数と一致させることが推奨されます。

3. Kafkaのインストール

ただ理論を語っても無駄だ。 Centos7 を例に、kafka のインストールと使用方法を紹介します。

Kafka では、サービス インスタンスのメタデータを保存するために zookeeper が必要なので、kafka をインストールする前に、まず zookeeper をインストールする必要があります。

3.1. ZooKeeperをインストールする

Zookeeperのインストール環境はjdkに依存しているため、事前にjdkをインストールする必要があります。

 # jdk1.8をインストール
yum -y java-1.8.0-openjdk をインストールします

Zookeeperをダウンロードし、ファイルパッケージを解凍します。

 #zookeeperをオンラインでダウンロード
http://mirrors.hust.edu.cn/apache/zookeeper/zookeeper-3.4.12/zookeeper-3.4.12.tar.gz を取得します。 
 
 #解凍
tar -zxvf zookeeper-3.4.12.tar.gz

データとログのディレクトリを作成する

#データとログを保存するためのディレクトリを作成する
cd /usr/zookeeper/
 mkdirデータ
mkdir ログ
 
 #conf の下にある zoo_sample.cfg ファイルをバックアップし、名前を zoo.cfg に変更します。
 cd conf/
 cp zoo_sample.cfg zoo.cfg

ZooKeeper を構成する

#zoo.cfgファイルを編集する
vim zoo.cfg

dataDirとdataLogDirのストレージパスを再構成する

最後に、Zookeeperサービスを開始します

#Zookeeperのbinディレクトリに入る
cd zookeeper/zookeeper-3.4.12/bin 
 
 #Zookeeper を起動する
./zkServer.sh 開始
 
 #Zookeeper のステータスを照会する
./zkServer.sh ステータス
 
 #飼育係のステータスを閉じる
./zkServer.sh 停止

3.2. Kafkaをインストールする

公式 Web サイト http://kafka.apache.org/downloads.html にアクセスして、必要なバージョンをダウンロードしてください。最新の安定バージョン 2.8.0 をダウンロードしました。

 #kafkaインストールパッケージをダウンロードする
https://apache.osuosl.org/kafka/2.8.0/kafka-2.8.0-src.tgz を取得します。 
 
 # ファイルパッケージを解凍する
tar -xvf kafka-2.8.0-src.tgz

必要に応じて構成ファイル server.properties を変更します (オプション)

 #設定フォルダに入る
kafka-2.8.0-src/config をコピーします
 
 #server.propertiesを編集する
vim サーバーのプロパティ

server.properties ファイルの内容は次のとおりです。

ブローカーID=0
リスナー=プレーンテキスト://localhost:9092
ネットワークスレッド数=3
スレッド数=8
ソケット送信バッファバイト数=102400
ソケット受信バッファバイト数=102400
ソケットリクエスト最大バイト数 = 104857600
 log.dirs=/tmp/kafka-logs
パーティション数=1
データディレクトリあたりの回復スレッド数=1
オフセット.トピック.レプリケーション.係数=1
トランザクション状態ログレプリケーション係数=1
トランザクション.state.log。最小.isr=1
ログ保持時間=168
ログセグメントバイト=1073741824
 log.retention.check.interval.ms =300000
 zookeeper.connect =ローカルホスト:2181
動物園の飼育員。接続.timeout.ms=6000
グループ.initial.rebalance.delay.ms=0

重要なパラメータは 4 つあります。

• broker.id: 一意のID
• listeners=PLAINTEXT://localhost:9092: kafka サービスのリスニング アドレスとポート
• log.dirs: ログ保存ディレクトリ
• zookeeper.connect: zookeeperサービスのアドレスを指定します

必要に応じて対応する構成を変更できます。

3.3. Kafkaサービスを開始する

# binスクリプトディレクトリに入る
kafka-2.8.0-src/bin をコピーします

Kafkaサービスを開始する

nohup kafka-server-start.sh ../config/server.properties server.log 2> server.err &

3.4.トピックを作成する

パーティションが 1 つだけ含まれ、レプリカが 1 つだけある testTopic というトピックを作成します。

 # binスクリプトディレクトリに入る
kafka-2.8.0-src/bin をコピーします
 
 #トピックを作成する
kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testTopic  

トピックを表示するには、list topic コマンドを実行します。

 # binスクリプトディレクトリに入る
kafka-2.8.0-src/bin をコピーします
 
 #現在のkafkaのすべてのトピックをクエリする
kafka-topics.sh --list --zookeeper ローカルホスト:2181  

出力：

テストトピック

3.5 メッセージの送信

Kafka には、ファイルまたは標準入力から入力を受け取り、それをメッセージとして Kafka クラスターに送信するコマンドライン クライアントが付属しています。デフォルトでは、各行は個別のメッセージとして送信されます。

プロデューサーを実行し、コンソールにメッセージを入力してサーバーに送信します。

 # binスクリプトディレクトリに入る
kafka-2.8.0-src/bin をコピーします
 
 #プロデューサーを実行し、testTopicトピックにメッセージを送信します
kafka-console-producer.sh --broker-list localhost:9092 --topic テストトピック 

2 つの項目を入力して Enter キーを押します。

こんにちは、カフカ！
これはメッセージです

3.5.メッセージの受信

Kafka には、メッセージを標準出力にダンプするコマンドライン コンシューマーもあります。

 # binスクリプトディレクトリに入る
kafka-2.8.0-src/bin をコピーします
 
 #testTopic トピックからメッセージをプルするコンシューマーを実行します
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testTopic --from-beginning  

出力は次のようになります。

こんにちは、カフカ！
これはメッセージです

IV.まとめ

この記事では主に、Kafka のアーキテクチャ モデルとインストール環境について予備的な概要を説明します。誤解が生じるのは避けられません。ネットユーザーの皆様は批判や苦情を歓迎します。

スペースの制約により、Java 環境での Kafka の適用シナリオについては、次の記事で詳しく紹介します。