分散フレームワークを読んで、知っておくべきNIOの基本的な知識

[[397718]]

この記事はWeChatの公開アカウント「KK Architect」から転載したもので、著者はKK Architectです。記事の転載についてはKKアーキテクトの公式アカウントまでご連絡ください。

1. はじめに

分散オープンソースプロジェクトを読むときに最も重要なことは、プロジェクトの通信フレームワークを理解することです。

分散オープンソースフレームワークは通常、クラスターにデプロイされるため、複雑な機能を実行するには異なるノードが相互に通信する必要があります。ソースコードを読むときに、その通信の仕組みを理解していないと、まるで原始の森に迷い込んだかのように、コードの中で迷子になってしまいます。

たとえば、HDFS は独自のカプセル化された Hadoop Rpc 通信フレームワークを使用します。 Spark は基盤となる通信に Netty を使用します。そして、私が最近読んだ Kafka のソースコードでは、最下層でネイティブ Java NIO が使用されています。

そこで今回は、Java NIO の主な知識ポイントについてお話ししましょう。

2. 複数の図でNIOの3つのコアコンセプトを理解する

NIO について話すとき、チャネル、バッファ、セレクターという 3 つのコア概念があります。

いきなり本題に入ると少し混乱するかもしれないので、最初から始める必要があります。

1. チャンネル

過去、同時実行要件がそれほど高くなかったときは、次に示すように、CPU がすべての入力と出力 (割り込み) の処理を担当していました。

ユーザープログラムはサーバーへの読み取りおよび書き込み要求を開始し、CPU はこれらの要求を直接処理します。これには欠点があります。 IO 要求が大量に発生すると、CPU が大量に占有され、システム全体の処理能力が低下します。

コンピュータの発展に伴い、次に示すように、DMA を使用して IO 要求を完全に処理する新しい方法が登場しました。

DMA は Direct Memory Access、直接メモリアクセス制御の略です。

なぜこのデバイスを追加する必要があるのですか?これは、CPU 割り込みモードがデータ転送速度の要件を満たすことができないためです。割り込みモードでは、割り込みごとにブレークポイントとシーンを保存する必要があり、割り込みが戻ったときにブレークポイントとシーンを復元する必要があります。

これらすべての理由により、割り込み方式では高速周辺機器の伝送速度要件を満たすことが困難になります。

そのため、DMA のようなデバイスが存在します。 DMA モードでのデータ転送プロセス中に、I/O デバイスがデータを転送する必要がある場合、DMA コントローラを介して CPU に DMA 転送要求を行います。 CPU が応答すると、システムバスを放棄し、DMA コントローラがバスを引き継いでデータ転送を行います。このとき、CPU は一部の初期化操作を除いて独自の処理を実行できます。

しかし、DMA を使用しても、I/O 要求が多すぎるとバスの競合が発生するため、急速なビジネス開発のニーズを満たすことはできません。

それでチャンネルは後から登場したのです。 DMA との違いは、チャネルが独自の命令システムとプログラムを持ち、コプロセッサであることです。 DMA は固定のデータ転送制御のみを実装できます。

Java NIO のチャネルは、上図のチャネルの実装です。

2. バッファ

チャネルの概念を理解すれば、バッファも簡単に理解できます。

チャネルは、I/O デバイス (ファイル、ソケットなど) へのオープン接続を表しますが、チャネル自体はデータを保存しません。バッファはデータ転送の実際のキャリアです。

アプリケーションがデータを書き込む場合、まずデータをバッファに書き込む必要があります。その後、チャネルはバッファ内のデータを宛先 (ファイル、ディスク、ネットワーク) に送信し、バッファからデータを取り出す役割を担います。

NIO システムを使用する場合は、I/O デバイスに接続するためのチャネルとデータを収容するためのバッファを取得し、バッファを操作してデータを処理する必要があります。

3. セレクター

セレクターはマルチプレクサーとも呼ばれ、非ブロッキング I/O です。非ブロッキングについて話しているうちに、まずブロッキングについて話す必要があります。ブロッキング方法を次の図に示します。

クライアントがサーバーに読み取りまたは書き込み要求を送信すると、サーバーのスレッドはカーネルアドレス空間にデータがあるかどうかを継続的にチェックします。

クライアントに送信するデータがない場合、サーバースレッドは待機し続け、この期間中は何も実行できません。

クライアントがデータを送信するまで、データはカーネルアドレス空間からユーザーアドレス空間にコピーされ、その後、データを読み取ることができます。

つまり、大量のリクエストが来た場合、後続のリクエストは前のリクエストの実行を待たなければならず、多くのキューが発生し、CPU リソースを十分に活用できず、パフォーマンスが急激に低下します。

セレクターがどのように機能するかを見てみましょう。

これで、クライアントとサーバー間の通信ではチャネル + バッファが使用されるため、すべてのチャネルがセレクタに登録されます。セレクターは、接続、読み取り、書き込みなど、これらのチャネルの I/O ステータスを監視します。

チャネル上のイベントが完全に準備されると、セレクターはタスクをサーバー上の 1 つ以上のスレッドに割り当てます。

クライアントにイベントの準備ができていない場合、サーバースレッドはブロックされません。クライアントイベントの準備ができるまで、独自の処理を実行できます。

ブロッキング方式と比較して、この非ブロッキング方式では CPU リソースをさらに活用できます。

3. 概念を理解してからAPIを学ぶ

1. バッファAPI

バッファを完全に理解するには、バッファの 4 つのプロパティ (マーク、位置、制限、容量) を知っておく必要があります。これらを知るには、コードを一度実行するだけです。

（１）一定サイズのバッファを割り当てる

//1.指定されたサイズのバッファを割り当てる
ByteBuffer バッファ = ByteBuffer.allocate(10);
システム。 out .println( "--------割り当て" );
システム。出力.println( "位置:" + buffer.position());
システム。出力.println( "limit:" + buffer.limit());
システム。出力.println( "容量:" + buffer.capacity());

操作結果:

 ---------割り当て------------  
位置:0
制限:10
定員:10

ここでは 10 バイトのバッファを割り当てます。つまり、最後の byte[] hb に 10 バイトのスペースを開きます。 ByteBuffer のプロパティ。

つまり、容量は 10、制限はデータの読み書きが可能な最大位置も 10、位置はデータの操作が可能な位置は 0 です。

（２）バッファにデータを書き込む

// 2. バッファにデータを書き込む
文字列 str = "abcde" ;
システム。出力.println( "------------put-------------" );
 buffer.put(str.getBytes(StandardCharsets.UTF_8));
システム。出力.println( "位置:" + buffer.position());
システム。出力.println( "limit:" + buffer.limit());
システム。出力.println( "容量:" + buffer.capacity());

操作結果:

 -  -  -  -  -  - 置く -  -  -  -  -  -  
位置:5
制限:10
定員:10

ここでは 5 バイトのデータをバッファに書き込むため、容量と制限は依然として 10 ですが、すでに 5 バイトが書き込まれているため、位置は 5 になります。

（３）データ読み取りモードに切り替える

// 3. データ読み取りモードに切り替える
バッファを反転します。
システム。出力.println( "-------------反転----------------" );
システム。出力.println( "位置:" + buffer.position());
システム。出力.println( "limit:" + buffer.limit());
システム。出力.println( "容量:" + buffer.capacity());

ここで、バッファからデータを読み取る場合は、フリップモードに切り替える必要があります。反転すると、いくつかの属性の値が変更されます。

操作結果:

 ------------- 反転-------------  
位置:0
制限:5
定員:10

flipは位置の値を0に変更し、制限を5に変更します。つまり、最初から読み取りたいのですが、位置5までしか読み取れません。

（4）データを読む

// 4. データの読み取り
システム。出力.println( "-------------get----------------" );
 byte[] dest = 新しいbyte[buffer.limit()];
バッファを取得します。
システム。出力.println(新しい文字列(dest,0,dest.length));
システム。出力.println( "位置:" + buffer.position());
システム。出力.println( "limit:" + buffer.limit());
システム。出力.println( "容量:" + buffer.capacity());

操作結果:

 -  -  -  -  -  - 得る -  -  -  -  -  -  
 ABCD
位置:5
制限:5
定員:10

データを読み取った後、位置は 5 になり、5 を読み取ったことを示します。

（５）繰り返し読む

//5.巻き戻し()
バッファを巻き戻す();
システム。出力.println( "-------------巻き戻し----------------" );
システム。出力.println( "位置:" + buffer.position());
システム。出力.println( "limit:" + buffer.limit());
システム。出力.println( "容量:" + buffer.capacity());

操作結果:

 ------------巻き戻し-------------  
位置:0
制限:5
定員:10

巻き戻しとは、バッファ内のデータを繰り返し読み取ることを意味する。

（６）データの消去

//6.クリア()
バッファをクリアします。
システム。出力.println( "--------------クリア----------------" );
システム。出力.println( "位置:" + buffer.position());
システム。出力.println( "limit:" + buffer.limit());
システム。出力.println( "容量:" + buffer.capacity());

操作結果:

 -  -  -  -  -  - クリア -  -  -  -  -  -  
位置:0
制限:10
定員:10

clear() 後、position は 0 に戻り、limit は 10 に戻り、最大 10 バイトまでデータを先頭から再度書き込むことができます。

ただし、バッファ内のデータはクリアされず、データはまだバッファ内に残っており、「忘れられた」状態にあることに注意してください。これらのポインタは元の状態に戻ります。

（７）マーキング

これは 4 番目の属性である「マーク」です。

マークはポジションの位置を記録することができます。 reset() メソッドを使用してマーク位置に戻ることができます。

 @テスト
パブリックボイドtest2() {
        // 10バイトを割り当てる
        文字列 str = "abcde" ;
        ByteBuffer バッファ = ByteBuffer.allocate(10);
        buffer.put(str.getBytes(StandardCharsets.UTF_8)); 
 
        // 読み取りモードに切り替えて2バイト読み取ります
        バッファを反転します。
        byte[] dest = 新しいbyte[buffer.limit()];
        バッファを取得します(宛先、0、2);
        システム。出力.println(新しいString(dest, 0, 2));
        システム。出力.println(buffer.position()); 
 
        // 現在の位置をマークする
        バッファをマークします。 
 
        // さらに2バイトを読み込む
        バッファを取得します(宛先, 2, 2);
        システム。出力.println(new String(dest, 2, 2));
        システム。出力.println(buffer.position()); 
 
        // リセット、マーク位置に戻る
        バッファをリセットします。
        システム。出力.println(buffer.position());
    } 
 
実行結果: 
 
 ```TeX
アブ
2
 CD
 4
 2

2. チャネル、バッファ、セレクタを使用してネットワークプログラムを完成させる

（１）サーバー

@テスト
パブリックvoid testServer()はIOExceptionをスローします{
     ServerSocketChannel serverSocketChannel = ServerSocketChannel。開ける（）;
     serverSocketChannel.configureBlocking( false ); 
 
     serverSocketChannel.bind(新しいInetSocketAddress(8989)); 
 
     セレクター セレクター = セレクター。開ける（）;
     serverSocketChannel.register(セレクタ、SelectionKey.OP_ACCEPT); 
 
     while (セレクタ.select () > 0) {
         イテレータ<SelectionKey> iterator = selector.selectedKeys().iterator();
         (イテレータ.hasNext()) の間 {
             選択キーキー= iterator.next ( );
             if (キー.isAcceptable()) {
                 ソケットチャネル socketChannel = serverSocketChannel.accept();
                 socketChannel.configureBlocking( false );
                 socketChannel.register(セレクタ、SelectionKey.OP_READ);
             }それ以外の場合 (キー.isReadable()) {
                 SocketChannel チャネル = (SocketChannel)キー.channel();
                 バイトバッファ byteBuffer = ByteBuffer.allocate(1024);
長さ = 0;
                 while ((len = channel. read (byteBuffer)) > 0) {
                     バイトバッファを反転します。
                     システム。出力.println(新しいString(byteBuffer.array(), 0, len));
                     バイトバッファをクリアします。
                 }
             }
         } 
 
         イテレータを削除します。
     }
 }

1. まず、ServerSocketChannel.open() を使用してチャネルを開き、非ブロッキングモードに設定します。

2. ポート8989にバインドします。

3. セレクターにチャネルを登録します。

4. while ループで、セレクターにイベントがあるかどうかを確認します。イベントがクライアントの接続イベントである場合は、SocketChannel を開き、非ブロッキングモードで登録し、セレクターにデータ読み取りイベントを登録します。

5. クライアントがデータを送信すると、チャネルを開いてバッファ内のデータを読み取ることができます。

6. このとき、サーバーは複数のクライアントからのリクエストを同時に受け入れることができます。

（２）クライアント

@テスト
パブリックvoid testClient()はIOExceptionをスローします{
     SocketChannel socketChannel = SocketChannel。新しいInetSocketAddress( "127.0.0.1" , 8989)を開きます。
     socketChannel.configureBlocking( false ); 
 
     バイトバッファ byteBuffer = ByteBuffer.allocate(1024);
     byteBuffer.put(新しいDate ().toString().getBytes(StandardCharsets.UTF_8));
     バイトバッファを反転します。
     socketChannel.write(byteBuffer);
     バイトバッファをクリアします。 
 
     socketChannel.close () ; 
 
 }