Anarchy In the 1K

Java NIOを用てスケーラブルI/Oを実現する

Java NIO ネットワーク 技術調査

目次

背景

　先日、こちら記事でスケーラブルI/Oの実現方法を調査しました。結果として、1つのスレッドで複数のI/Oを管理することが必要であり、その実現に以下が必要だと分かりました。

　今回は、上記をJavaで実現する方法を調査します。Java NIOを用いたサンプル実装に関して、後日公開を行う予定です。

ノンブロッキングI/Oの実装方法

　ストリームI/OはノンブロッキングI/Oに対応してない為、今回はNIO(New IO)を用いた実装を行います。

NIOとストリームI/Oの比較

　従来のストリームI/Oでは、入出力先とのデータパスを、ストリームオブジェクトとして表現しています。そして、それらをつなぎ合わせることで、以下の様にバッファリングやデータ変換機能を追加します。

Socket socket = new Socket("localhost", 8080);
InputStream inputStream = socket.getInputStream();
// バッファリング機能の追加
InputStream bufferedInputStream = new BufferedInputStream(inputStream);
// プリミティブ型へのデータ変換機能の追加
DataInputStream dataInputStream = new DataInputStream(bufferedInputStream);

　一方、NIOでは、以下の通りオブジェクト毎に役割が分担されています。

  • Channel: 入出力先とのデータパスを表すオブジェクトです。
  • Buffer: バッファリングとデータ変換機能を提供するオブジェクトです。

両者の関係は下図の通りです。
f:id:fujiU:20200327201610p:plain

Bufferクラス

　Bufferクラスに関して、仕組みが複雑である為、以下に大まかな仕様をまとめます。

概要

　Bufferは、ある1つの種類のプリミティブ型を格納可能なオブジェクトです。要素の取り込み, 取り出しの2つの用途に使用可能です。

主な属性

  • capacity: Bufferに格納可能な要素数の上限を表します。
  • position: 要素の取り込み, 取り出しの開始位置を表します。
  • limit: 要素の取り込み, 取り出しが行えない範囲の先頭の位置を表します。

主な操作

  • putメソッド: Bufferへ要素の取り込みを行うメソッドです。
  • getメソッド: Bufferから要素の取り出しを行うメソッドです。
  • flipメソッド: Bufferの属性であるpositionとlimitを、下図の通り"取り込み時の値"と"取り出し時の値"に相互に切り替えを行うメソッドです。 f:id:fujiU:20200327231727p:plain
  • compactメソッド: 下図の通り、以下2つの操作を行うメソッドです。
    1. 既に取り出した要素を捨て、残っている要素をBufferの先頭に移動する。
    2. Bufferの属性であるpositionとlimitを、"取り出し時の値"から"取り込み時の値に"に切り替える。

    f:id:fujiU:20200329171710p:plain

I/Oの多重化の実装方法

利用するクラス一覧

　以下、3つのクラスを利用します。

  • SelectableChannel: Selectorに対応したChannelを表すクラスです。
  • SelectionKey: SelectableChannelをSelectorに登録した際の戻り値であり、両者のひも付きを表すクラスです。
  • Selector: 目的の状態になったChannelを取得する為のクラスです。

これらのクラスの関係は、下図の通りです。SelectableChannelを複数のSelectorに登録可能であり、登録毎にSelectionKeyが発行されます。 f:id:fujiU:20200329193028p:plain
　Selectorを介して、目的の状態を満たしたSelectableChannelを取得することで、1つのスレッドで複数の接続が管理可能になります。

SelectableChannelをSelectorに登録する方法

　SelectableChannelクラスのregisterメソッドを用いて、以下の通りSelectorにSelectableChannelを登録します。

selectableChannel.register(selector, SelectionKey.OP_xxx);

第2引数として、目的の状態を表すのビットパターンの集合を渡します。該当のビットパターンに関して、SelectionKeyにおいて、以下の通り定義されています。

public static final int OP_READ = 1 << 0; // 00001
public static final int OP_WRITE = 1 << 2; // 00100
public static final int OP_CONNECT = 1 << 3; // 01000
public static final int OP_ACCEPT = 1 << 4; // 10000

目的の状態を複数登録する場合は、以下の通り行います。(|はOR演算子です。)

selectableChannel.register(selector,
    SelectionKey.OP_READ | SelectionKey.OP_WRITE); // 101