Kafka ネットワーク層ソースコード実装メカニズムにおけるメッセージの送受信プロセス全体の図解説明

みなさんこんにちは。Hua Zaiです。またお会いできて嬉しいです。

今日は主に Kafka ネットワーク層の送受信プロセスをまとめます。このシリーズは3つの記事に分かれています。これは次の記事で、主に最後の質問を分析します。

1. Java NIO の SocketChannel の場合、Kafka はどのようにして統合トランスポート層をカプセル化し、最も基本的なネットワーク接続と読み取りおよび書き込み操作を実装するのでしょうか?
2. KafkaChannel がトランスポート層をカプセル化し、バッファ操作の読み取りと書き込みを行う方法を分析します。
3. 産業グレードの NIO 実践の分析: ビット操作に基づいてイベント監視を制御する方法と、アンパックとスティッキングを実装する方法。
4. Kafka がセレクター マルチプレクサーをカプセル化する方法を分析します。
5. Kafka によってカプセル化されたセレクターがどのように初期化され、ブローカーに接続され、ネットワークの読み取りと書き込みを実行するかを分析します。
6. Kafka ネットワークでメッセージを送信し、応答を受信するプロセス全体とは何ですか?

この記事を注意深く読むと、Kafka ネットワーク層のソースコードに対する理解が深まると思います。

この記事には役立つ情報が沢山含まれているので、じっくり読んでいただければ幸いです。

I. 全体概要

シナリオ駆動型のアプローチを通じて、ネットワーク要求がカプセル化され監視された後、ネットワーク経由でメッセージがどのように送受信されるか、そして何を行う必要があるかを見てみましょう。

1. メッセージ送信プロセスの分析

• メッセージの事前送信
• メッセージは実際に送信されます

2. 受信応答プロセス分析

• 回答結果を読む
• 応答情報の解析
• コールバックの処理

誰もが理解しやすいように、すべてのソースコードの骨組みのみが保持されます。

2. メッセージ送信プロセスの分析

1. メッセージの事前送信

この部分は多くのことが関係しているので、ここでは簡単に説明し、後でそれを分析する特別な章があります。

クライアントはまず、送信するメッセージを準備します。プロセスは次のとおりです。

1. Sender 子スレッドは、送信するメッセージ セットを RecordAccumulator バッファーから取得し、抽出されたデータは次の場所に保存されます。

• 送信時に、inFlightRequests コレクションと KafkaChannel の send オブジェクトが追加されます。 inFlightRequests については、後続の章で分析します。簡単に説明します。このコレクションは、送信するメッセージのバッファ領域を保存および操作するために使用されます。リクエストがネットワーク経由で送信される準備が整うと、リクエストはキューの先頭からキューに入れられます。応答が受信されると、リクエストはキューの末尾から削除されます。
• 送信が完了すると、completedRequests コレクションに格納されます。

2. 期限切れデータの処理。
3. クライアント要求 ClientRequest をカプセル化し、ClientRequest クラス オブジェクトを NetworkClient に送信します。主に以下の 2 つのタスクを実行します。

• ClientRequest クラス オブジェクトに基づいて InFlightRequest クラス オブジェクトを構築します。
• ClientRequest クラス オブジェクトに基づいて NetworkSend クラス オブジェクトを構築し、それを KafkaChannel キャッシュに格納します。

4. この時点でメッセージの事前送信は終了します。

次に、Selector クラスと KafkaChannel クラスの具体的なソース コード実装を順番に見ていきましょう。

（１）メモリに一時的に保存するデータを要求する

github ソースコードのアドレスは次のとおりです。

https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/common/network/Selector.java

https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java

 /**
 * メッセージの事前送信
*/
 パブリックvoid send ( 送信送信) {
 // 1. サーバーからconnectionIdを取得する
文字列connectionId = 送信します。 行き先（）;
 // 2. データパケットから対応する接続​​を取得する
KafkaChannel チャネル= openOrClosingChannelOrFail ( connectionId );
 // 3. 閉じた接続セット内に接続が存在する場合
if ( closechannels.containsKey ( connectionId )) {
 // connectionId を failedSends コレクションに格納する
これ。 送信に失敗しました。 ( 接続ID ) を追加します。
 } それ以外{
 試す{
 // 4. 一時データは事前​​に送信されますが、実際には送信されません。一度に送信できるのは 1 つだけです
チャンネル。 setSend ( 送信);
 } キャッチ( 例外e ) {
 // 5. KafkaChannel のステータスを更新して失敗を送信する
チャンネル。 状態( ChannelState . FAILED_SEND );
 // 6. connectionIdをfailedSendsコレクションに格納する
これ。 送信に失敗しました。 ( 接続ID ) を追加します。
 // 7. 接続を閉じる
close ( チャネル、 CloseMode 。 DISCARD_NO_NOTIFY ) ;
 ...
 }
 }
 }

ソースコードから、KafkaChannel クラスの setSend() メソッドが呼び出されていることがわかります。

 パブリックvoid setSend ( 送信送信) {
 if ( this . send != null )
 throw new IllegalStateException ( "前の送信操作がまだ進行中の状態で送信操作を開始しようとしました。接続 ID は " + idです);
 // 送信するメッセージのフィールドを設定する
これ。 送信= 送信;
 //トランスポート層を呼び出して書き込みイベントを追加します
これ。 トランスポートレイヤー。 addInterestOps ( 選択キー. OP_WRITE );
 }

 // PlaintextTransportLayer クラスのメソッド
@オーバーライド
パブリックvoid addInterestOps ( int ops ) {
 //key.interestOps() 経由でイベントを追加します |オプス
鍵。 interestOps ( キー.interestOps () | ops );
 }

このメソッドは主に事前送信に使用されます。つまり、ネットワーク要求を送信する前に、送信する ByteBuffer データが KafkaChannel の send に保存され、その後トランスポート層メソッドが呼び出されて、このチャネル上の "OP_WRITE" イベントへの注目度が高まり、同時に "OP_READ" イベントも保持されます。このとき、チャネルは読み取りと書き込みを同時に行うことができます。実際に送信が実行されると、まず送信からデータが読み取られます。

2. メッセージが実際に送信される

Sender 子スレッドは、Selector の「poll」メソッドを呼び出して、実際にリクエストを送信します。

(1) 投票()

 @オーバーライド
パブリックvoid ポーリング( 長いタイムアウト) はIOException をスローします{
 ...
 // nioSelector.select を呼び出してスレッドをブロックし、I/O イベントを待機し、ブロック時間を設定し、I/O イベントの準備ができるまで待機し、監視された準備完了イベントの数を返します。
 int numReadyKeys = select ( タイムアウト);
 // コレクションが空でないかキャッシュされたデータが存在する場合は、イベントをリッスンするか、すぐに接続します
if ( numReadyKeys > 0 || ! immediatelyConnectedKeys .isEmpty () || dataInBuffers ) {
 // キャッシュデータはSSL接続でのみ存在する可能性があります
if ( データインバッファ) {
 // イベントを処理する
pollSelectionKeys ( toPoll 、 false 、 endSelect );
 }
 // 監視された準備完了イベントを処理する
pollSelectionKeys ( readyKeys 、 false 、 endSelect );
 // 即時接続コレクションを処理する
pollSelectionKeys ( immediateConnectedKeys 、 true 、 endSelect );
 } それ以外{
 ...
 }
 ...
 }

このメソッドは、準備完了イベントを収集し、そのイベントに対してネットワーク操作を実行するという 1 つの処理を実行します。上記の簡略化されたコードから、「pollSelectionKeys」メソッドが呼び出されていることがわかります。実際の読み取りおよび書き込み操作はこのメソッドで行われます。見てみましょう:

(2) ポーリング選択キー()

 void pollSelectionKeys ( <SelectionKey> selectionKeys を設定し、 boolean isImmediatelyConnected 、 long currentTimeNanos を設定します) {
 //1.現在監視されているイベントをループします（元の順序またはシャッフルされた順序で）
 for ( SelectionKey キー: determineHandlingOrder ( selectionKeys )) {
 // 2. 事前に接続を作成し、kafkachanelをキーに登録します。ここで対応するチャネルを取得します。
 KafkaChannel チャネル= チャネル( キー);
 ...
 // 3. ノードIDを取得する
文字列nodeId = チャネル。 id ();
 ...
 試す{
 ...
 // 4. イベントの準備ができているかどうかを読み取る
if ( channel.ready ( ) && ( key.isReadable () || channel.hasBytesBuffered ()) && ! hasCompletedReceive ( channel ) && ! exticallyMutedChannels.contains ( channel ) ) {
 // 読み取りイベントを処理してみる
attemptRead ( チャネル);
 }
 ...
 試す{
 // 5. 書き込みイベントの処理を試みる
書き込みを試行します( キー、 チャネル、 nowNanos )。
 } キャッチ( 例外e ) {
 送信失敗= true ;
 e を投げる;
 }
 } キャッチ( 例外e ) {
 ...
 ついに
 ....
 }
 }
 }

このメソッドは主に、接続イベント、読み取りおよび書き込みイベント、すぐに完了する接続など、監視対象イベントを処理するために使用されます。次に、実際にネットワークに書き込む方法を見てみましょう。

(3) 書き込みを試みる()

 プライベートvoid attemptWrite （ SelectionKey キー、 KafkaChannel チャネル、 long nowNanos ）はIOException をスローします{
 // ここで、書き込み操作を実行する前に 4 つの条件を満たす必要があります
if ( チャネル.hasSend ()
 && チャンネル。 準備ができて（）
 ＆＆ 鍵。 書き込み可能()
 && ! チャンネル。 多分クライアント再認証の開始(() -> nowNanos )) {
 //書き込み操作を実行する
書き込み( チャネル);
 }
 }
 // チャネル接続準備完了
パブリックブール値 ready () {
 transportLayer を返します。 ready () && 認証子。 完了（）;
 }

 // java nio 選択キー
パブリックファイナルブールisWritable () {
 戻り値( readyOps () & OP_WRITE ) != 0 ;
 }

このメソッドは主に、ネットワーク書き込み操作を実行するために使用されます。この方法は非常にシンプルで、「同時に 4 つの条件を満たす」必要があります。

1. 「チャネルにはまだ送信するデータがあります」は、データがまだ送信されていないことを意味します。
2. 「チャネル接続準備完了。」
3. 「書き込みイベントは書き込み可能な状態です」書き込みバッファがいっぱいでない限り、「 OP_WRITE 」イベントは生成され続けます。データが書き込まれていない場合、またはデータがいっぱいの場合は、不要なリソース消費を防ぐために、「 OP_WRITE 」イベントをキャンセルする必要があります。
4. 「クライアント認証が有効になっていません。」

上記の 4 つの条件が満たされると、書き込み操作を実行できます。次に、書き込み操作のプロセスを見てみましょう。

(4) 書き込み()

 // 書き込み操作を実行する
void write ( KafkaChannel チャネル) はIOException をスローします{
 // 1. チャネルに対応するノードIDを取得する
文字列nodeId = チャネル。 id ();
 // 2. sendに格納されたデータは実際に送信されますが、一度にすべて送信されない可能性があり、送信されたバイト数が返されます。
 長いバイト送信= チャネル。 書く（）;
 // 3. 送信が完了したかどうかを判断します。そうでない場合はnullを返し、次のポーリングが送信を続行するまで待ちます。
 send = チャネルを送信します。 おそらく完了送信();
 // 4. メッセージが送信または完了したことを示します
送信バイト数が0 以上のとき、 送信はnull になります。
 長いcurrentTimeMs = 時間。 ミリ秒（）;
 送信バイト数> 0の場合
//送信バイトメトリック情報を記録する
これ。 センサー。 送信バイト数を記録します( ノード ID 、 送信バイト数、 現在の送信時間)。
 // 送信完了
if ( 送信!= null ) {
 // 送信をcompletedSendsに追加する
これ。完了しました。 送信します。 追加（ 送信）;
 //送信された完了したメトリック情報を記録する
これ。 センサー。 recordCompletedSend ( nodeId 、 send 、 size ()、 currentTimeMs );
 }
 }
 }

このメソッドは主に、ネットワーク書き込み操作を実際に実行するために使用されます。ご存知のとおり、ネットワーク プログラミングのプロセスでは、送信を 1 回で完了できるとは限りません。このとき、送信が完了したかどうかを判断する必要があります。そうでない場合は、null が返され、「送信を継続し、このチャネルの書き込みイベントに注意を払い続けるために、次のポーリング poll() を待機します」。送信が完了したら、「送信を返し、このソケットチャネルの OP_WRITE イベントに対するセレクターの注意をキャンセルします。」ここでは、書き込み操作を実行するために KafkaChannel クラスの write() メソッドが呼び出され、送信が完了したかどうかを判断するために maybeCompleteSend() が呼び出されます。まず write() 操作を見てみましょう。

(5) KafkaChannel.write()

 パブリックlong write ()はIOException をスローします{
 // 送信が空かどうかを確認します。空の場合は、メッセージが送信されたことを意味します。
 if ( 送信== null )
 0 を返します。
 書き込みがtrueの場合;
 // 実際にデータを送信するには ByteBufferSend.writeTo を呼び出します
返送します。 writeTo ( トランスポートレイヤー);
 }

このメソッドは主に、send に保存されたデータを実際に送信するために使用されます。実際のデータを送信するには、ByteBufferSend.writeTo を呼び出します。 writeTo() メソッドを見てみましょう:

 @オーバーライド
// バイトストリームデータをチャネルに書き込む
パブリックlong writeTo ( GatheringByteChannel チャネル) はIOException をスローします{
 // 1. nio の基になる write メソッドを呼び出して、トランスポート層にバッファを書き込み、書き込まれたバイト数を返します。
 長く書き込まれた= チャネル。 書き込み( バッファ);
 （ 書かれていない< 0 ） の場合
throw new EOFException ( "負のバイトをチャネルに書き込みました。これは発生しないはずです。" );
 // 2. トランスポート層に書き込む残りのバイト数を計算する
残り-= 書き込まれました;
 // 送信するたびにチェックする
保留中= TransportLayers 。 hasPendingWrites ( チャネル);
 返却書面;
 }

このメソッドは主に、バッファ配列を SocketChannel に書き込むために使用されます。ネットワーク プログラミングでは、一度書き込むだけではすべてのデータが正常に書き込まれない場合があるため、Java NIO の基盤となる channel.write(buffers) メソッドを呼び出すと、「正常に書き込まれたバイト数」の戻り値が返され、一度呼び出した後に書き込まれたバイト数を知ることができます。

書き込み操作のために write() と一連の基礎となるメソッドが呼び出されると、送信されたバイト数が返されます。今回送信が完了していない場合は、null が返され、「ネットワーク書き込み操作の送信を継続するために次のポーリングを待機し、このチャネルの書き込みイベントに注意を払い続けます」ため、今回送信が完了したかどうかを判断する必要があります。見てみましょう:

(6) 送信完了()

 //送信を完了できる
パブリック送信maybeCompleteSend () {
 // 送信は空ではなく、送信済みです
if ( send != null && send .completed ()) {
 中間書き込み= false ;
 // データが書き込まれたら、トランスポート層の OP_WRITE イベントの監視をキャンセルし、書き込み操作を完了します。
 トランスポートレイヤー。 関心操作を削除します( SelectionKey.OP_WRITE );
 // 結果セットの結果に送信を割り当てる
結果を送信= send ;
 // 読み取った後、次回書き込めるように送信をクリアします
送信= null ;
 //最後に結果セットの結果を返して書き込み操作を完了します
結果を返します。
 }
 null を返します。
 }
 // PlaintextTransportLayer クラスのメソッド
@オーバーライド
パブリックvoid removeInterestOps ( int ops ) {
 // key.interestOps() と ~ops でイベントを削除する
鍵。 interestOps ( キー.interestOps () & ~ops );
 }
 // バイトバッファ送信
@オーバーライド
パブリックブール値完了（）{
 残り<= 0 を返す&& ! 保留中;
 }

この方法は主にデータの書き込みが完了したかどうかを判断するために使用されます。データの書き込みが完了したかどうかを判断する条件は、バッファに何も残っておらず、pending が false であることです。送信が完了したら、完了したリクエストを完了した送信コレクションcompletedSendsに追加します。

メッセージ要求が送信された後、他に何が行われますか?これには、NetworkClient クラスに関する関連知識が必要です。以下に簡単な説明と、さらに詳しい分析を示します。

github ソースコードのアドレスは次のとおりです。

https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

 private void handleCompletedSends ( < ClientResponse > 応答のリスト、 現在長い) {
 // 応答が期待されない場合は送信が完了したらそれを返します
// 送信後、sendをcompletedSendsコレクションに追加し、このコレクションを走査します
for ( Send send : this.selector.completedSends ()) {
 // inFlightRequests コレクションから対応するブローカーに送信された最後のリクエスト要素を取得します
InFlightRequest リクエスト= this 。 フライトリクエスト。 lastSent ( 送信. 送信先());
 // 応答が期待されるかどうかを決定する
if ( ! request . expectResponse ) {
 // 応答が期待されない場合は、対応するブローカー要求キューに送信された inFlightRequests コレクションの最初の要素を削除します
これ。 フライトリクエスト。 completeLastSent ( 送信. 送信先() );
 // リクエストをレスポンスコレクションに追加する
応答。 追加( リクエスト. 完了( null 、 現在) );
 }
 }
 }

ソース コードから、「completedSends」コレクションと「inFlightRequests」コレクションには「相互に連携する」関係があることがわかります。

「completedSends」コレクションは送信されたがまだ返されていないリクエスト コレクションを参照し、「inFlightRequests」コレクションは送信されたが応答結果を受け取っていないリクエスト コレクションを格納します。 「completedSends」の要素は、「inFlightRequests」コレクションに対応するキューの最後の要素に対応します。

これでメッセージ送信プロセスの分析は終了です。送信後の後続作業については、Sender と NetWorkClient の説明時に詳しく分析します。次に、レスポンスの受信処理を見てみましょう。

3. 受信応答プロセスの分析

​

上記の Selector.pollSelectionKeys() を分析すると、ネットワーク読み取りイベントの準備ができると、ネットワーク読み取り操作を試行するために attemptRead() が呼び出されます。見てみましょう:

1. 回答結果を読む

(1) 試行読み取り()

 プライベートvoid attemptRead ( KafkaChannel チャネル) はIOException をスローします{
 // チャネルに対応するノードIDを取得します
文字列nodeId = チャネル。 id ();
 // トランスポート層からNetworkReceiveオブジェクトにデータを読み込む
長いバイト受信= チャネル。 読む（）;
 受信バイト数が0の場合
 ...
 // NetworkReceiveオブジェクトが読み取られたかどうかを判定する
NetworkReceive 受信= チャネル。 たぶん完了受信();
 // 読み取った後、このNetworkReceiveオブジェクトを受信したネットワーク要求のコレクションに追加します
受信がnull の場合
addToCompletedReceives ( チャネル、 受信、 現在のタイムミリ秒);
 }
 }
 ...
 }
 // KafkaChannel メソッド
パブリックlong read ()はIOException をスローします{
 受信がnull の場合
// NetworkReceive オブジェクトを初期化する
受信= 新しいNetworkReceive ( maxReceiveSize 、 id 、 memoryPool );
 }
 // チャネルデータをNetworkReceiveオブジェクトに読み込んでみる
long bytesReceived = 受信( this . 受信);
 ...
 受信したバイトを返します。
 }

このメソッドは主に、データを読み取り、受信したコレクションに追加するために使用されます。最初に KafkaChannel.read() メソッドが呼び出されて読み取りが行われ、その後読み取りが完了したかどうかが判断されることがわかります。そうでない場合は、次のポーリングで読み取りを続行します。完了した場合は、リクエストのcompletedReceivesコレクションに追加されます。

NetworkReceive オブジェクトが読み取られたかどうかを確認する方法を見てみましょう。

(2) 多分完了受信()

 // NetworkReceiveオブジェクトが読み取られたかどうかを判定する
// この時点で NetworkReceive オブジェクト全体が読み取られていない場合は、次に読み取りイベントがトリガーされたときに、NetworkReceive オブジェクト全体が引き続き読み取られます。
 // 完全な NetworkReceive オブジェクトが読み取られると、null に設定されます。次に読み取りイベントがトリガーされると、新しい NetworkReceive オブジェクトが作成されます。
 パブリックNetworkReceive maybeCompleteReceive () {
 if ( receive != null && receive .complete ()) {
 受け取る。 ペイロード()。 巻き戻し（）;
 NetworkReceive 結果= 受信;
 受信= null ;
 結果を返します。
 }
 null を返します。
 }
 // ネットワーク受信
パブリックブール値完全（）{
 戻る！ サイズ。 hasRemaining () && バッファ!= null && ! バッファ。 残りがある();
 }

このメソッドは主に、データが読み取られたかどうかを判断するために使用されます。読み取りが完了したかどうかを判断する条件は、応答メッセージ ヘッダーのサイズ ByteBuffer と応答メッセージ本体のバッファー ByteBuffer を含め、NetworkReceive 内のバッファーが使い果たされたかどうかです。両方読んだときに初めて読書は完了します。

この時点で NetworkReceive オブジェクト全体が読み取られていない場合は、次に読み取りイベントがトリガーされたときに、NetworkReceive オブジェクト全体が引き続き読み取られます。この時点で完全な NetworkReceive オブジェクトが読み取られている場合は、そのオブジェクトはクリアされ、次に読み取りイベントがトリガーされたときに新しい NetworkReceive オブジェクトが作成されます。

2. 応答メッセージを解析する

完全な応答メッセージを読んだ後、次に何をすべきでしょうか?つまり、応答メッセージを解析します。やり方を見てみましょう:

github ソースコードのアドレスは次のとおりです。

https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

 private void handleCompletedReceives ( < ClientResponse > 応答をリストします。 現在は長いです) {
 // 読み取り後、このNetworkReceiveオブジェクトを受信したネットワーク要求のコレクションに追加し、このコレクションを走査します
for ( NetworkReceive 受信: this . selector . completeReceives ()) {
 // リクエストを送信したノードIDを取得します
文字列ソース= 受信。 ソース（）;
 // InFlightRequestコレクションから対応する要素を取得して削除します
InFlightRequest 要求= inFlightRequests 。 completeNext ( ソース);
 // レスポンスを解析する
構造体responseStruct = parseStructMaybeUpdateThrottleTimeMetrics ( 受信. ペイロード()、 要求. ヘッダー、
 throttleTimeSensor 、 現在);
 ....
 // レスポンスをレスポンス結果セットに追加する
応答。 追加( 要求。 完了( 応答、 現在) )
 }
 }

このメソッドは主に、completedReceives コレクションをループして応答処理を実行するために使用されます。記事の冒頭で簡単に述べたように、応答を受信した後、応答は「inFlightRequests」から削除され、その後応答が解析されます。

 プライベート静的構造体parseStructMaybeUpdateThrottleTimeMetrics ( ByteBuffer responseBuffer 、 RequestHeader requestHeader 、 センサーthrottleTimeSensor 、 long now ) {
 // レスポンスヘッダーを取得する
レスポンスヘッダレスポンスヘッダ= レスポンスヘッダ。 解析( responseBuffer 、 requestHeader . apiKey () 。 responseHeaderVersion ( requestHeader . apiVersion ()));
 // レスポンス本文を取得する
構造体responseBody = requestHeader 。 apiKey () 。 parseResponse ( requestHeader . apiVersion ()、 responseBuffer );
 // レスポンス ヘッダーの correlationId とレスポンス本文の correlationId を比較して、それらが一致しているかどうかを確認します。一致していない場合は例外をスローします。
 相関関係( requestHeader 、 responseHeader );
 ...
 レスポンスボディを返します。
 }

このメソッドは主に、応答を解析し、応答ヘッダーの correlationId 値が応答本体の値と一致しているかどうかを判断するために使用されます。一致しない場合は例外がスローされます。

現時点では、応答は解析されるだけで、処理されません。応答処理は、コールバック メソッドを呼び出すことによって処理されます。見てみましょう。

3. コールバックを処理する

 private void completeResponses ( < ClientResponse > 応答のリスト) {
 // レスポンス結果セットを走査する
( ClientResponse レスポンス: レスポンス) {
 試す{
 応答。 完了();
 } キャッチ( 例外e ) {
 ログ。 error ( "リクエストの完了時に捕捉されないエラー:" , e );
 }
 }
 }
 //ClientResponse クラス
パブリックvoid onComplete () {
 if ( コールバック!= null )
 折り返し電話。 onComplete ( これ);
 }

これで、応答メッセージを受信するプロセスの分析が完了します。

IV.結論

ここで、この記事の要点をまとめてみましょう。

1. まず、Kafka ネットワーク層の送受信プロセス全体を見てみましょう。これは主に「メッセージ送信プロセス」と「応答受信プロセス」に分かれています。

2. メッセージ送信プロセスと応答受信プロセスのソースコード実装の詳細も分析します。