この記事を読むまで私はカフカのことをよく知っていると思っていた

Kafka は、もともと LinkedIn で開発されたメッセージング システムであり、LinkedIn のアクティビティ ストリームと運用データ処理パイプラインの基盤として使用されています。

[[273550]]

画像はPexelsより

現在、さまざまな種類の企業で、複数の種類のデータ パイプラインやメッセージング システムとして使用されています。アクティビティ ストリーム データは、ほぼすべてのサイトがサイトの使用状況を報告するときに使用する最も一般的なデータです。

アクティビティ データには、ページ ビュー、閲覧したコンテンツに関する情報、検索結果が含まれます。

この種のデータを処理する通常の方法は、まずさまざまなアクティビティをログの形式で何らかのファイルに書き込み、次にこれらのファイルに対して定期的に統計分析を実行することです。

運用データとは、サーバーのパフォーマンス データ (CPU、IO 使用率、要求時間、サービス ログなど) を指します。運用データにはさまざまな統計手法があります。

近年、キャンペーンおよび運用データの処理は、Web サイト ソフトウェア製品機能の重要なコンポーネントとなっており、それをサポートするには、やや複雑なインフラストラクチャが必要です。

Kafka の基本概念

Kafka は、次の主な設計目標を持つ、分散型のパブリッシュ/サブスクライブ ベースのメッセージング システムです。

• これは、O(1) の時間計算量でメッセージの永続性を提供し、TB レベルを超えるデータに対しても一定の時間計算量のアクセス パフォーマンスを保証できます。
• 高いスループット。非常に安価な商用マシンでも、1 台のマシンで 1 秒あたり 10 万件を超えるメッセージの送信をサポートできます。
• 各パーティション内でのメッセージの順次送信を保証しながら、Kafka サーバー間のメッセージのパーティション分割と分散消費をサポートします。
• オフラインデータ処理とリアルタイムデータ処理の両方をサポートします。
• スケールアウト: オンラインでの水平拡張をサポートします。

生産者と消費者

Kafka には、基本的に 2 種類のクライアントがあります。

• プロデューサー
• 消費者

さらに、データ統合用の Kafka Connect API やストリーム処理用の Kafka Streams などの高レベル クライアントもありますが、これらの高レベル クライアントの基盤となるレイヤーは依然としてプロデューサー API とコンシューマー API であり、上位レイヤーでカプセル化されているだけです。

これは簡単に理解できます。プロデューサー (パブリッシャーとも呼ばれます) がメッセージを作成し、コンシューマー (サブスクライバーとも呼ばれます) がメッセージを消費または読み取る責任を負います。

トピックとパーティション

Kafka では、メッセージはトピック別に分類され、各トピックはデータベース内のテーブルに似た「メッセージ キュー」に対応します。

ただし、類似のメッセージをすべて「中央」キューに入れると、必然的にスケーラビリティが欠如することになります。プロデューサー/コンシューマーの数が増えたり、メッセージの数が増えたりすると、システムのパフォーマンスやストレージが枯渇する可能性があります。

実際の例を使って説明しましょう。都市 A で生産された商品を道路で都市 B に輸送する必要があります。

したがって、単一チャネルの高速道路では、「都市 A の貨物量が増加する」か、「都市 C も都市 B に貨物を輸送する必要がある」かに関係なく、「スループットが不十分」という問題が生じます。

そこで、私たちは、テーマの水平方向の拡大を実現するために、「より多くの道路を建設できるようにする」ことに似たパーティションの概念を導入します。

ブローカーとクラスター

Kafka サーバーはブローカーとも呼ばれます。プロデューサーから送信されたメッセージを受け入れ、ディスクに保存します。ブローカーは、パーティション メッセージをプルするというコンシューマーの要求にも応え、これまでに送信されたメッセージを返します。

ブローカーは特定のマシン ハードウェアを使用して、1 秒あたり数万のパーティションと数百万のメッセージを処理できます。 （最近は数百万になることも多いです。調べてみたところ、クラスタリングの場合は確かにスループットはかなり高いようです。）

複数のブローカーがクラスターを形成し、クラスター内のブローカーがクラスター コントローラーとなり、ブローカーへのパーティションの割り当てやブローカーの障害の監視など、クラスターの管理を担当します。

クラスターでは、パーティションはブローカーによって管理されます。ブローカーはパーティションのリーダーとも呼ばれます。

もちろん、パーティションを複数のブローカーに複製して冗長性を実現できるため、ブローカーに障害が発生した場合、そのパーティションを他のブローカーに再割り当てして処理させることができます。

次の図は例です。

Kafka の重要な特性はログの保持です。ログを一定期間のみ保持したり、特定のサイズのログのみ保持するなど、トピックのメッセージ保持ポリシーを設定できます。

これらの制限を超えると、古いメッセージは削除されます。特定のトピックに対して個別のメッセージ有効期限ポリシーを設定して、さまざまなアプリケーションに合わせてカスタマイズすることもできます。

複数のクラスター

ビジネスが成長するにつれて、通常は次のような理由から複数のクラスターが必要になることがよくあります。

• データに基づく分離
• セキュリティベースの分離
• 複数のデータセンター（災害復旧）

複数のデータセンターを構築する場合、メッセージの相互通信を実現することが必要になることがよくあります。たとえば、ユーザーが個人情報を変更した場合、どのデータセンターがそれを処理するかに関係なく、この更新を後続のリクエストに反映する必要があります。あるいは、データ分析のために、複数のデータセンターからのデータをマスターコントロールセンターに集約する必要があります。

上記のパーティション レプリケーション冗長性メカニズムは、同じ Kafka クラスター内でのみ適用されます。複数の Kafka クラスター間でメッセージを同期するには、Kafka が提供する MirrorMaker ツールを使用できます。

本質的に、MirrorMaker はキューを使用して接続された Kafka のコンシューマーとプロデューサーにすぎません。あるクラスターからのメッセージを消費し、別のクラスターへのメッセージを生成します。

Kafkaの設計と実装

上記では Kafka の基本的な概念をいくつか学習しましたが、成熟した「メッセージ キュー」ミドルウェアとして、検討する価値のある興味深い設計が数多くあります。以下にそのいくつかを簡単に挙げてみましょう。

Kafkaはファイルシステムに保存されます

はい、まず Kafka メッセージがファイル システム上に存在することを知っておく必要があります。 Kafka は、メッセージを保存およびキャッシュするためにファイル システムに大きく依存しています。一般的に人々は「ディスクは遅い」と考えており、そのような設計には懐疑的です。

実際には、ディスクは使用方法に応じて、人々が期待するよりもはるかに速くなったり、はるかに遅くなったりします。適切に設計されたディスク構造により、ネットワークと同等の速度を実現できます。

最新のオペレーティング システムでは、ディスク アクセスを高速化するために、ディスクの読み取りと書き込みに対していくつかの最適化が行われています。

たとえば、先読みでは、比較的大きなディスクを事前にメモリに読み込みます。ライトビハインドは、多数の小さな論理書き込み操作を 1 つの大きな物理書き込み操作に結合します。

さらに、オペレーティング システムはメイン メモリ内の残りのすべての空きメモリ領域をディスク キャッシュとして使用し、すべてのディスク読み取りおよび書き込み操作は統合ディスク キャッシュを経由します (ディスク キャッシュをバイパスする直接 I/O を除く)。

これらの最適化機能を組み合わせると、ディスクへのシーケンシャル アクセスの場合、ランダム メモリ アクセスよりも高速になる場合があり、ネットワークの速度とほぼ同じになることもあります。

上記のトピックは、実際には消費者と生産者に向けた論理的な概念です。物理的に保存されるのは実際にはパーティションです。各パーティションは最終的にディレクトリに対応し、すべてのメッセージとインデックス ファイルが格納されます。

デフォルトでは、トピックの作成時にパーティションの数が指定されていない場合は、パーティションが 1 つだけ作成されます。

たとえば、test という名前のトピックを作成し、パーティションの数を指定しないと、デフォルトで test-0 フォルダーが作成されます。ここでの命名規則は次のとおりです。 - 。

パーティションに公開されたメッセージは、パーティション データ ファイルの末尾に追加されます。このシーケンシャルなディスク書き込み操作により、Kafka は非常に効率的になります (シーケンシャルなディスク書き込みはランダムなメモリ書き込みよりも効率的であることが証明されており、これは Kafka の高スループットを保証する非常に重要なものです)。

各メッセージはブローカーに送信され、パーティション ルールに従って保存されるパーティションが選択されます。パーティション ルールが適切に設定されていれば、すべてのメッセージを異なるパーティションに均等に分散できます。

Kafka の基盤となるストレージ設計

Kafka クラスターにはブローカーが 1 つだけあると仮定します。それぞれ 1 個と 2 個のパーティションを持つ、「Topic1」と「Topic2」という名前の 2 つのトピックを作成します。

すると、ルート ディレクトリに次の 3 つのフォルダーが作成されます。

 | --トピック1-0  
 | --トピック2-0  
 | --トピック2-1  

Kafka のファイル ストレージでは、同じトピックの下に複数の異なるパーティションがあり、各パーティションはディレクトリです。

各ディレクトリは、同じサイズの複数のセグメント ファイルに均等に分割されます。セグメント ファイルは、インデックス ファイルとデータ ファイルで構成されます。常にペアで表示されます。サフィックス「.index」と「.log」は、セグメント インデックス ファイルとデータ ファイルを表します。

ここで、各セグメントのサイズを 500 MB に設定し、プロデューサーが大量のデータを topic1 に書き込むように開始するとします。すると、topic1-0 フォルダーに次のようなファイルが生成されます。

 | --トピック1-0  
    | --000000000000000000000.インデックス 
    | --000000000000000000000.log  
    | --00000000000000368769.インデックス 
    | --00000000000000368769.log  
    | --00000000000000737337.インデックス 
    | --00000000000000737337.log  
    | --00000000000001105814.インデックス 
    | --00000000000001105814.log  
 | --トピック2-0  
 | --トピック2-1  

セグメントは、Kafka ファイル ストレージの最小単位です。セグメント ファイルの命名規則: グローバル パーティションの最初のセグメントは 0 から始まり、後続の各セグメント ファイル名は、前のセグメント ファイルの最後のメッセージのオフセット値になります。

最大値は 64 ビット長、19 桁、0 で埋められていない任意の桁です。たとえば、000000000000000368769.index や 00000000000000368769.log などです。

上記のセグメント ファイルのペアを例に、インデックス ファイルとデータ ファイルの対応関係について説明します。

インデックス ファイル内のメタデータ <3, 497> を例にとると、これはデータ ファイル内の 3 番目のメッセージ (グローバル パーティション内の 368769 + 3 = 368772 番目のメッセージ) を表し、メッセージの物理オフセット アドレスは 497 です。

インデックス ファイルは 0 から始まるわけではなく、毎回 1 ずつ増加するわけでもないことに注意してください。これは、Kafka がスパース インデックス ストレージを使用し、一定数のデータごとにインデックスを作成するためです。

これにより、インデックス ファイルのサイズが縮小され、インデックスをメモリにマップできるようになり、クエリ中のディスク IO オーバーヘッドが削減され、クエリにかかる時間が短縮されます。

ファイル名は前のセグメントの最後のメッセージのオフセットであるため、指定されたオフセットを持つメッセージを検索する必要がある場合は、すべてのセグメントのファイル名でバイナリ検索を実行することで、そのメッセージが属するセグメントを見つけることができます。

次に、インデックス ファイル内でファイルに対応する物理的な場所を見つけると、メッセージを取得できます。

メッセージはパーティションのセグメント データ ファイル内で順次読み書きされ、消費後に削除されないため (削除ポリシーは期限切れのセグメント ファイル用)、これがシーケンシャル ディスク IO ストレージ デザイナー Kafka が高パフォーマンスである重要な理由です。

Kafka はどのようにしてメッセージの正確なオフセットを知るのでしょうか?これは、Kafka が標準のデータ ストレージ構造を定義しているためです。パーティション内の各メッセージには、次の 3 つの属性が含まれます。

• オフセット: 現在のパーティション内のメッセージのオフセットを示します。これは、パーティション内のメッセージを一意に識別する論理値であり、単純に ID として考えることができます。
• MessageSize: メッセージ コンテンツ データのサイズを示します。
• データ: メッセージの具体的な内容。

プロデューサーデザイン概要

メッセージを送信する前に、いくつかの質問をします。各メッセージは重要であり、失われてはなりませんか?時々メッセージが重複しても大丈夫ですか?書き込まれたメッセージのメッセージ遅延またはスループットについて懸念がありますか?

たとえば、トランザクションが発生したときに Kafka にメッセージを送信するクレジットカードトランザクション処理システムがあります。別のサービスがメッセージを読み取り、ルール エンジンに基づいてトランザクションが成功したかどうかを確認し、結果を Kafka を通じて返します。

このようなビジネスでは、メッセージが失われたり重複したりすることは許されません。トランザクション量が多いため、スループットをできるだけ大きくする必要があり、レイテンシが若干高くなる可能性があります。

別の例として、Web ページ上のユーザーのクリック データを収集する必要があるとします。このようなシナリオでは、少量のメッセージの損失や重複は許容され、ユーザー エクスペリエンスに影響を与えない限り、遅延は重要ではありません。スループットはリアルタイム ユーザーの数によって決まります。

ビジネスによって必要な書き方や構成は異なります。ここでは具体的な方法については説明しません。次に、プロデューサーがメッセージを書き込む基本的なプロセスを見てみましょう。

プロセスは次のとおりです。

• まず、ProducerRecord を作成する必要があります。これには、メッセージのトピックと値が含まれる必要があり、オプションでキーまたはパーティションを指定できます。
• メッセージを送信する際、プロデューサーはキーと値をバイト配列にシリアル化し、ディストリビューター (パーティショナー) に送信します。
• パーティションを指定すると、アロケータはそのパーティションを返します。それ以外の場合、アロケータはキー値に基づいてパーティションを選択し、それを返します。
• パーティションを選択すると、プロデューサーはメッセージが属するトピックとパーティションを認識します。このレコードを同じトピックとパーティションのバッチ メッセージに追加します。別のスレッドが、これらのバッチ メッセージを対応する Kafka ブローカーに送信する役割を担います。
• ブローカーがメッセージを受信すると、正常に書き込まれた場合は、メッセージのトピック、パーティション、および変位を含む RecordMetadata オブジェクトを返します。それ以外の場合は例外を返します。
• プロデューサーは結果を受け取った後、例外があれば再試行することがあります。

消費者向けデザイン概要

① 消費者及び消費者団体

このようなシナリオを想定します。Kafka からメッセージを読み取り、チェックし、最終的に結果データを生成します。

これを行うにはコンシューマー インスタンスを作成することもできますが、プロデューサーがメッセージを書き込む速度がコンシューマーがメッセージを読み取る速度よりも速い場合はどうなるでしょうか。

時間が経つにつれて、メッセージの山はますます深刻になります。このシナリオでは、水平拡張のために複数のコンシューマーを追加する必要があります。

Kafka コンシューマーはコンシューマー グループの一部です。複数のコンシューマーがトピックを消費するためにコンシューマー グループを形成する場合、各コンシューマーは異なるパーティションからメッセージを受信します。

4 つのパーティションを持つ T1 トピックがあるとします。同時に、消費者グループ G1 があり、このグループには消費者 C1 が 1 つだけあります。

次に、コンシューマー C1 は、以下に示すように、これらの 4 つのパーティションからメッセージを受信します。

新しいコンシューマー C2 をコンシューマー グループ G1 に追加すると、各コンシューマーは以下に示すように 2 つのパーティションからメッセージを受信します。

コンシューマーの数が 4 に増えると、以下に示すように、各コンシューマーは 1 つのパーティションからメッセージを受信します。

しかし、このコンシューマー グループにコンシューマーを追加し続けると、残りのコンシューマーはアイドル状態になり、メッセージを受信しなくなります。

まとめると、消費者グループ内の消費者数を増やして水平展開を行うことで、消費能力を高めることができます。

そのため、トピックを作成するときは、より多くのパーティションを使用して、消費負荷が高いときにコンシューマーを追加してパフォーマンスを向上させることをお勧めします。

さらに、余分なコンシューマーはアイドル状態になり、何の役にも立たないため、コンシューマーの数はパーティションの数より大きくしないでください。

Kafka の非常に重要な機能は、メッセージを一度だけ書き込むだけでよく、任意の数のアプリケーションがメッセージを読み取ることができることです。

つまり、すべてのアプリケーションがメッセージの全量を読み取ることができます。各アプリケーションがメッセージの全量を読み取るためには、アプリケーションに異なるコンシューマー グループが必要です。

上記の例で、新しいコンシューマー グループ G2 を追加し、このコンシューマー グループに 2 つのコンシューマーがある場合、次のようになります。

このシナリオでは、コンシューマー グループ G1 とコンシューマー グループ G2 の両方がトピック T1 からの全量のメッセージを受信できます。論理的に言えば、それらは異なるアプリケーションに属します。

最後にまとめると、アプリケーションがメッセージの全量を読み取る必要がある場合は、アプリケーションのコンシューマー グループを設定してください。アプリケーションの消費容量が不十分な場合は、このコンシューマー グループにコンシューマーを追加することを検討できます。

②消費者グループとパーティションのバランスを再調整する

ご覧のとおり、新しいコンシューマーがコンシューマー グループに参加すると、以前に他のコンシューマーによって消費されていた 1 つ以上のパーティションを消費します。

さらに、コンシューマーがコンシューマー グループを離れると (たとえば、再起動、ダウンタイムなどにより)、そのコンシューマーが消費するパーティションは他のパーティションに割り当てられます。

この現象はリバランスと呼ばれます。リバランスは Kafka の非常に重要な特性であり、高可用性と水平スケーラビリティを保証します。

ただし、再バランス調整期間中はすべてのコンシューマーがメッセージを消費できないため、コンシューマー グループ全体が一時的に使用できなくなることに注意してください。

さらに、パーティションを再調整すると、元のコンシューマーの状態が期限切れになり、コンシューマーが状態を再更新することになり、この期間中のコンシューマーのパフォーマンスも低下します。後ほど、リバランスを安全に実行する方法と、可能であればそれを回避する方法について説明します。

コンシューマーは、グループ コーディネーターとして機能するブローカーにハートビートを定期的に送信することで、コンシューマー グループ内で存続します。

このブローカーは固定されておらず、消費者グループごとに異なる場合があります。コンシューマーがメッセージをプルまたはコミットすると、ハートビートが送信されます。

コンシューマーが一定期間ハートビートを送信しない場合、そのセッションは期限切れとなり、グループ コーディネーターはコンシューマーがクラッシュしたと想定して、再バランスをトリガーします。

コンシューマーのクラッシュからセッションの有効期限が切れるまでに一定の時間があり、その間コンシューマーのパーティションはメッセージを消費できないことがわかります。

通常、コンシューマーがグループ コーディネータに離脱メッセージを送信し、グループ コーディネータがセッションの有効期限が切れるのを待たずにすぐに再バランス調整できるように、エレガントなシャットダウンを実行できます。

バージョン 0.10.1 では、Kafka はハートビートのメカニズムを変更し、ハートビートの送信とメッセージのプルを分離しました。これにより、ハートビートの送信頻度はメッセージのプル頻度の影響を受けなくなりました。

さらに、Kafka のより高いバージョンでは、メッセージをプルせずにコンシューマーが存続できる期間を構成することがサポートされています。この構成により、ライブロックを回避できます。ライブロックとは、アプリケーションに欠陥はないが、何らかの理由によりそれ以上使用できないことを意味します。

③パーティションと消費モデル

前述のように、Kafka のトピック内のメッセージは複数のパーティションに分散して保存されます。コンシューマー グループは、消費時に異なるパーティションからメッセージを取得する必要があります。では、最終的にトピック内のメッセージの順序をどのように再構築するのでしょうか?

答えは、「方法はありません」です。 Kafka は、グローバルな状況に関係なく、メッセージがパーティション内で順序付けられていることのみを保証します。

次の質問は、パーティション内のメッセージは複数回（異なるコンシューマー グループによって）消費される可能性があるので、パーティション内の消費されたメッセージはいつ削除されるのかということです。パーティションはどのようにしてコンシューマー グループの現在の消費状況を把握するのでしょうか?

メッセージが消費されたかどうかに関係なく、パーティションは期限が切れない限りメッセージを削除することはありません。たとえば、保持期間が 2 日に設定されている場合、どのグループもメッセージが公開されてから 2 日以内にそのメッセージを消費できます。 2日後、メッセージは自動的に削除されます。

パーティションは各コンシューマー グループのオフセットを保存し、グループによって消費された位置を記録します。以下のように表示されます。

④Kafkaがプルモデルである理由

コンシューマーはブローカーにデータを要求すべきか (プル)、ブローカーはコンシューマーにデータをプッシュすべきか (プッシュ)?

メッセージング システムとして、Kafka はプロデューサーがブローカーにメッセージをプッシュし、コンシューマーがブローカーからメッセージをプルするという従来のアプローチに従います。

Facebook の Scribe や Cloudera の Flume などの一部のログ中心のシステムは、プッシュ モデルを使用します。実際、プッシュ モードとプル モードにはそれぞれ長所と短所があります。

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

プッシュ モードの目的は、できるだけ早くメッセージを配信することですが、これにより、コンシューマーがメッセージを処理する時間が十分になくなる可能性が高まり、通常はサービス拒否やネットワークの輻輳が発生します。

プル モードでは、コンシューマーの消費容量に基づいて適切な速度でメッセージを消費できます。

Kafka の場合、Pull モードの方が適しています。プル モードではブローカーの設計を簡素化でき、コンシューマーはメッセージの消費速度を独立して制御できます。

同時に、コンシューマーは消費モード、つまりバッチ消費または行ごとの消費を独自に制御でき、異なる送信セマンティクスを実現するために異なる送信方法を選択することもできます。

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

信頼性について議論するときは、必ず「保証」という言葉を使います*。信頼性の保証は、私たちがアプリケーションを構築する基盤となります。

たとえば、リレーショナル データベースの信頼性の保証は ACID です。ACID は、原子性、一貫性、独立性、永続性を表します。

Kafka における信頼性の保証は次のとおりです。

• パーティションの場合、メッセージは順序付けられます。プロデューサーがパーティションにメッセージ A を書き込んでからメッセージ B を書き込むと、コンシューマーは最初にメッセージ A を読み取り、次にメッセージ B を読み取ります。
• メッセージは、同期中のすべてのレプリカに書き込まれた時点でコミットされたとみなされます。

ここでの書き込みはファイル システム キャッシュにのみ書き込まれ、ディスクにフラッシュされない可能性があります。プロデューサーは、パーティション マスター レプリカが書き込まれるのを待ってから戻るなど、さまざまなタイミングで確認を待つことができます。一方、プロデューサーは、同期状態のすべてのレプリカが書き込まれるのを待ってから戻ることができます。

• メッセージがコミットされると、少なくとも 1 つのレプリカが存続している限り、データは失われません。
• コンシューマーはコミットされたメッセージのみを読み取ることができます。

これらの基本的な保証を使用して、信頼性の高いシステムを構築します。この時点で、「アプリケーションはどの程度の信頼性が求められるのか」という質問について検討する必要があります。

信頼性は無料ではありません。これは、システムの可用性、スループット、レイテンシ、ハードウェアの価格に密接に関連しており、いずれか 1 つを失うことになります。したがって、トレードオフが必要になることが多く、信頼性を盲目的に追求することは現実的ではありません。

Kafkaを構築する

上記の説明により、Kafka が何であるかを大まかに理解できました。今、私たちはそれを体験するために地元でそれを構築しようとし始めています。

ステップ1: Kafkaをダウンロードする

ここでは Mac OS を例にとり、Homebrew がインストールされている場合に次のコードを実行します。

ブリューインストールkafka

Kafka は Zookeeper に依存しているため、ダウンロード中に自動的にダウンロードされます。

ステップ2: サービスを開始する

始める前に、まず Kafka のリスニング アドレスとポートを localhost:9092 に変更する必要があります。

 vi /usr/ local /etc/kafka/server.properties

次に、次のように変更します。

Zookeeper と Kafka を順番に起動します。

 BrewサービスがZookeeperを開始
brew サービスが kafka を起動します

次に、次のステートメントを実行して、「test」という名前のトピックを作成します。

 kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic テスト 

次のコマンドでトピック リストを表示できます。

 kafka-topics --list --zookeeper ローカルホスト:2181  

ステップ3: メッセージを送信する

次に、新しいコンソールを作成し、次のコマンドを実行して、作成したトピックをフォローするコンシューマーを作成します。

 kafka-console-consumer --bootstrap-server localhost:9092 --topic テスト --from-beginning  

コンソールを使用して、作成したトピックにメッセージを追加し、作成したコンシューマー ウィンドウを確認します。

 kafka-console-producer --broker-list localhost:9092 --topic テスト 

正しいメッセージは、消費者ウィンドウを通じて確認できます。

参考文献:

1. Kafka 設計分析 (I): Kafka の背景とアーキテクチャの紹介
2. カフカシリーズ（I）カフカを知る
3. カフカ入門
4. Kafka のトピックをパーティション分割する必要があるのはなぜですか? - 知湖
5. カフカのデザインと実践的思考
6. Kafka シリーズ (VI) 信頼性の高いデータ転送