Java中I/O模型的示例分析

這篇文章將為大家詳細講解有關Java中I/O模型的示例分析，小編覺得挺實用的，因此分享給大家做個參考，希望大家閱讀完這篇文章后可以有所收獲。

相關概念

同步和異步

描述的是用戶線程與內核的交互方式：

• 同步是指用戶線程發起 I/O 請求后需要等待或者輪詢內核 I/O 操作完成后才能繼續執行；

• 異步是指用戶線程發起 I/O 請求后仍繼續執行，當內核 I/O 操作完成后會通知用戶線程，或者調用用戶線程注冊的回調函數。

阻塞和非阻塞

描述的是用戶線程調用內核 I/O 操作的方式：

• 阻塞是指 I/O 操作需要徹底完成后才返回到用戶空間；

• 非阻塞是指 I/O 操作被調用后立即返回給用戶一個狀態值，無需等到 I/O 操作徹底完成。

一個 I/O 操作其實分成了兩個步驟：發起 I/O 請求和實際的 I/O 操作。 阻塞 I/O 和非阻塞 I/O 的區別在于第一步，發起 I/O 請求是否會被阻塞，如果阻塞直到完成那么就是傳統的阻塞 I/O ，如果不阻塞，那么就是非阻塞 I/O 。 同步 I/O 和異步 I/O 的區別就在于第二個步驟是否阻塞，如果實際的 I/O 讀寫阻塞請求進程，那么就是同步 I/O 。

Unix I/O 模型

Unix 下共有五種 I/O 模型：

• 阻塞 I/O

• 非阻塞 I/O

• I/O 復用（select 和 poll）

• 信號驅動 I/O（SIGIO）

• 異步 I/O（POSIX 的 aio_系列函數）

阻塞 I/O

請求無法立即完成則保持阻塞。

階段1：等待數據就緒。網絡 I/O 的情況就是等待遠端數據陸續抵達；磁盤I/O的情況就是等待磁盤數據從磁盤上讀取到內核態內存中。
階段2：數據從內核拷貝到進程。出于系統安全,用戶態的程序沒有權限直接讀取內核態內存,因此內核負責把內核態內存中的數據拷貝一份到用戶態內存中。

非阻塞 I/O

• socket 設置為 NONBLOCK（非阻塞）就是告訴內核，當所請求的 I/O 操作無法完成時，不要將進程睡眠，而是返回一個錯誤碼(EWOULDBLOCK) ，這樣請求就不會阻塞
  

• I/O 操作函數將不斷的測試數據是否已經準備好，如果沒有準備好，繼續測試，直到數據準備好為止。整個 I/O 請求的過程中，雖然用戶線程每次發起 I/O 請求后可以立即返回，但是為了等到數據，仍需要不斷地輪詢、重復請求，消耗了大量的 CPU 的資源
  

• 數據準備好了，從內核拷貝到用戶空間。

一般很少直接使用這種模型，而是在其他 I/O 模型中使用非阻塞 I/O 這一特性。這種方式對單個 I/O 請求意義不大,但給 I/O 多路復用鋪平了道路.

I/O 復用（異步阻塞 I/O）

I/O 多路復用會用到 select 或者 poll 函數，這兩個函數也會使進程阻塞，但是和阻塞 I/O 所不同的的，這兩個函數可以同時阻塞多個 I/O 操作。而且可以同時對多個讀操作，多個寫操作的 I/O 函數進行檢測，直到有數據可讀或可寫時，才真正調用 I/O 操作函數。

從流程上來看，使用 select 函數進行 I/O 請求和同步阻塞模型沒有太大的區別，甚至還多了添加監視 socket，以及調用 select 函數的額外操作，效率更差。但是，使用 select 以后最大的優勢是用戶可以在一個線程內同時處理多個 socket 的 I/O 請求。用戶可以注冊多個 socket，然后不斷地調用 select 讀取被激活的 socket，即可達到在同一個線程內同時處理多個 I/O 請求的目的。而在同步阻塞模型中，必須通過多線程的方式才能達到這個目的。

I/O 多路復用模型使用了 Reactor 設計模式實現了這一機制。

調用 select / poll 該方法由一個用戶態線程負責輪詢多個 socket,直到某個階段1的數據就緒,再通知實際的用戶線程執行階段2的拷貝。 通過一個專職的用戶態線程執行非阻塞I/O輪詢,模擬實現了階段一的異步化

信號驅動 I/O（SIGIO）

首先我們允許 socket 進行信號驅動 I/O,并安裝一個信號處理函數，進程繼續運行并不阻塞。當數據準備好時，進程會收到一個 SIGIO 信號，可以在信號處理函數中調用 I/O 操作函數處理數據。

異步 I/O

調用 aio_read 函數，告訴內核描述字，緩沖區指針，緩沖區大小，文件偏移以及通知的方式，然后立即返回。當內核將數據拷貝到緩沖區后，再通知應用程序。

異步 I/O 模型使用了 Proactor 設計模式實現了這一機制。

告知內核,當整個過程(包括階段1和階段2)全部完成時,通知應用程序來讀數據.

幾種 I/O 模型的比較

前四種模型的區別是階段1不相同，階段2基本相同，都是將數據從內核拷貝到調用者的緩沖區。而異步 I/O 的兩個階段都不同于前四個模型。

同步 I/O 操作引起請求進程阻塞，直到 I/O 操作完成。異步 I/O 操作不引起請求進程阻塞。

常見 Java I/O 模型

在了解了 UNIX 的 I/O 模型之后，其實 Java 的 I/O 模型也是類似。

“阻塞I/O”模式

在上一節 Socket 章節中的 EchoServer 就是一個簡單的阻塞 I/O 例子，服務器啟動后，等待客戶端連接。在客戶端連接服務器后，服務器就阻塞讀寫取數據流。

EchoServer 代碼：

public class EchoServer {
public static int DEFAULT_PORT = 7;

public static void main(String[] args) throws IOException {

int port;
try {
port = Integer.parseInt(args[0]);
} catch (RuntimeException ex) {
port = DEFAULT_PORT;
}
try (
ServerSocket serverSocket =
new ServerSocket(port);
Socket clientSocket = serverSocket.accept(); 
PrintWriter out =
new PrintWriter(clientSocket.getOutputStream(), true); 
BufferedReader in = new BufferedReader(
new InputStreamReader(clientSocket.getInputStream()));
) {
String inputLine;
while ((inputLine = in.readLine()) != null) {
out.println(inputLine);
}
} catch (IOException e) {
System.out.println("Exception caught when trying to listen on port "
+ port + " or listening for a connection");
System.out.println(e.getMessage());
}
}
}

改進為“阻塞I/O+多線程”模式

使用多線程來支持多個客戶端來訪問服務器。

主線程 MultiThreadEchoServer.java

public class MultiThreadEchoServer {
public static int DEFAULT_PORT = 7;
public static void main(String[] args) throws IOException {
int port;
try {
port = Integer.parseInt(args[0]);
} catch (RuntimeException ex) {
port = DEFAULT_PORT;
}
Socket clientSocket = null;
try (ServerSocket serverSocket = new ServerSocket(port);) {
while (true) {
clientSocket = serverSocket.accept();
// MultiThread
new Thread(new EchoServerHandler(clientSocket)).start();
}
} catch (IOException e) {
System.out.println(
"Exception caught when trying to listen on port " + port + " or listening for a connection");
System.out.println(e.getMessage());
}
}
}

處理器類 EchoServerHandler.java

public class EchoServerHandler implements Runnable {
private Socket clientSocket;
public EchoServerHandler(Socket clientSocket) {
this.clientSocket = clientSocket;
}
@Override
public void run() {
try (PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true);
BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));) {
String inputLine;
while ((inputLine = in.readLine()) != null) {
out.println(inputLine);
}
} catch (IOException e) {
System.out.println(e.getMessage());
}
}
}

存在問題：每次接收到新的連接都要新建一個線程，處理完成后銷毀線程，代價大。當有大量地短連接出現時，性能比較低。

改進為“阻塞I/O+線程池”模式

針對上面多線程的模型中，出現的線程重復創建、銷毀帶來的開銷，可以采用線程池來優化。每次接收到新連接后從池中取一個空閑線程進行處理，處理完成后再放回池中，重用線程避免了頻率地創建和銷毀線程帶來的開銷。

主線程 ThreadPoolEchoServer.java

public class ThreadPoolEchoServer {
public static int DEFAULT_PORT = 7;
public static void main(String[] args) throws IOException {
int port;
try {
port = Integer.parseInt(args[0]);
} catch (RuntimeException ex) {
port = DEFAULT_PORT;
}
ExecutorService threadPool = Executors.newFixedThreadPool(5);
Socket clientSocket = null;
try (ServerSocket serverSocket = new ServerSocket(port);) {
while (true) {
clientSocket = serverSocket.accept();
// Thread Pool
threadPool.submit(new Thread(new EchoServerHandler(clientSocket)));
}
} catch (IOException e) {
System.out.println(
"Exception caught when trying to listen on port " + port + " or listening for a connection");
System.out.println(e.getMessage());
}
}
}

存在問題：在大量短連接的場景中性能會有提升，因為不用每次都創建和銷毀線程，而是重用連接池中的線程。但在大量長連接的場景中，因為線程被連接長期占用，不需要頻繁地創建和銷毀線程，因而沒有什么優勢。

雖然這種方法可以適用于小到中度規模的客戶端的并發數，如果連接數超過 100,000或更多，那么性能將很不理想。

改進為“非阻塞I/O”模式

“阻塞I/O+線程池”網絡模型雖然比”阻塞I/O+多線程”網絡模型在性能方面有提升，但這兩種模型都存在一個共同的問題：讀和寫操作都是同步阻塞的,面對大并發（持續大量連接同時請求）的場景，需要消耗大量的線程來維持連接。CPU 在大量的線程之間頻繁切換，性能損耗很大。一旦單機的連接超過1萬，甚至達到幾萬的時候，服務器的性能會急劇下降。

而 NIO 的 Selector 卻很好地解決了這個問題，用主線程（一個線程或者是 CPU 個數的線程）保持住所有的連接，管理和讀取客戶端連接的數據，將讀取的數據交給后面的線程池處理，線程池處理完業務邏輯后，將結果交給主線程發送響應給客戶端，少量的線程就可以處理大量連接的請求。

Java NIO 由以下幾個核心部分組成：

• Channel

• Buffer

• Selector

要使用 Selector，得向 Selector 注冊 Channel，然后調用它的 select()方法。這個方法會一直阻塞到某個注冊的通道有事件就緒。一旦這個方法返回，線程就可以處理這些事件，事件的例子有如新連接進來，數據接收等。

主線程 NonBlokingEchoServer.java

public class NonBlokingEchoServer {
public static int DEFAULT_PORT = 7;
public static void main(String[] args) throws IOException {
int port;
try {
port = Integer.parseInt(args[0]);
} catch (RuntimeException ex) {
port = DEFAULT_PORT;
}
System.out.println("Listening for connections on port " + port);
ServerSocketChannel serverChannel;
Selector selector;
try {
serverChannel = ServerSocketChannel.open();
InetSocketAddress address = new InetSocketAddress(port);
serverChannel.bind(address);
serverChannel.configureBlocking(false);
selector = Selector.open();
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
} catch (IOException ex) {
ex.printStackTrace();
return;
}
while (true) {
try {
selector.select();
} catch (IOException ex) {
ex.printStackTrace();
break;
}
Set<SelectionKey> readyKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = readyKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
try {
if (key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel client = server.accept();
System.out.println("Accepted connection from " + client);
client.configureBlocking(false);
SelectionKey clientKey = client.register(selector,
SelectionKey.OP_WRITE | SelectionKey.OP_READ);
ByteBuffer buffer = ByteBuffer.allocate(100);
clientKey.attach(buffer);
}
if (key.isReadable()) {
SocketChannel client = (SocketChannel) key.channel();
ByteBuffer output = (ByteBuffer) key.attachment();
client.read(output);
}
if (key.isWritable()) {
SocketChannel client = (SocketChannel) key.channel();
ByteBuffer output = (ByteBuffer) key.attachment();
output.flip();
client.write(output);
output.compact();
}
} catch (IOException ex) {
key.cancel();
try {
key.channel().close();
} catch (IOException cex) {
}
}
}
}
}
}

改進為“異步I/O”模式

Java SE 7 版本之后，引入了異步 I/O （NIO.2） 的支持，為構建高性能的網絡應用提供了一個利器。

主線程 AsyncEchoServer.java

public class AsyncEchoServer {
public static int DEFAULT_PORT = 7;
public static void main(String[] args) throws IOException {
int port;
try {
port = Integer.parseInt(args[0]);
} catch (RuntimeException ex) {
port = DEFAULT_PORT;
}
ExecutorService taskExecutor = Executors.newCachedThreadPool(Executors.defaultThreadFactory());
// create asynchronous server socket channel bound to the default group
try (AsynchronousServerSocketChannel asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open()) {
if (asynchronousServerSocketChannel.isOpen()) {
// set some options
asynchronousServerSocketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024);
asynchronousServerSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
// bind the server socket channel to local address
asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
// display a waiting message while ... waiting clients
System.out.println("Waiting for connections ...");
while (true) {
Future<AsynchronousSocketChannel> asynchronousSocketChannelFuture = asynchronousServerSocketChannel
.accept();
try {
final AsynchronousSocketChannel asynchronousSocketChannel = asynchronousSocketChannelFuture
.get();
Callable<String> worker = new Callable<String>() {
@Override
public String call() throws Exception {
String host = asynchronousSocketChannel.getRemoteAddress().toString();
System.out.println("Incoming connection from: " + host);
final ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
// transmitting data
while (asynchronousSocketChannel.read(buffer).get() != -1) {
buffer.flip();
asynchronousSocketChannel.write(buffer).get();
if (buffer.hasRemaining()) {
buffer.compact();
} else {
buffer.clear();
}
}
asynchronousSocketChannel.close();
System.out.println(host + " was successfully served!");
return host;
}
};
taskExecutor.submit(worker);
} catch (InterruptedException | ExecutionException ex) {
System.err.println(ex);
System.err.println("\n Server is shutting down ...");
// this will make the executor accept no new threads
// and finish all existing threads in the queue
taskExecutor.shutdown();
// wait until all threads are finished
while (!taskExecutor.isTerminated()) {
}
break;
}
}
} else {
System.out.println("The asynchronous server-socket channel cannot be opened!");
}
} catch (IOException ex) {
System.err.println(ex);
}
}
}

關于“Java中I/O模型的示例分析”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，使各位可以學到更多知識，如果覺得文章不錯，請把它分享出去讓更多的人看到。