天天看點

Java NIO與阻塞IO的研究

一.java NIO 和阻塞I/O的差別 

1. 阻塞I/O通信模型

假如現在你對阻塞I/O已有了一定了解,我們知道阻塞I/O在調用InputStream.read()方法時是阻塞的,它會一直等到資料到來時(或逾時)才會傳回;同樣,在調用ServerSocket.accept()方法時,也會一直阻塞到有用戶端連接配接才會傳回,每個用戶端連接配接過來後,服務端都會啟動一個線程去處理該用戶端的請求。阻塞I/O的通信模型示意圖如下:

Java NIO與阻塞IO的研究

如果你細細分析,一定會發現阻塞I/O存在一些缺點。根據阻塞I/O通信模型,我總結了它的兩點缺點:

1. 當用戶端多時,會建立大量的處理線程。且每個線程都要占用棧空間和一些CPU時間

2. 阻塞可能帶來頻繁的上下文切換,且大部分上下文切換可能是無意義的。

在這種情況下非阻塞式I/O就有了它的應用前景。

2. java NIO原理及通信模型 

Java NIO是在jdk1.4開始使用的,它既可以說成“新I/O”,也可以說成非阻塞式I/O。下面是java NIO的工作原理:

1. 由一個專門的線程來處理所有的 IO 事件,并負責分發。 

2. 事件驅動機制:事件到的時候觸發,而不是同步的去監視事件。 

3. 線程通訊:線程之間通過 wait,notify 等方式通訊。保證每次上下文切換都是有意義的。減少無謂的線程切換。 

閱讀過一些資料之後,下面貼出我了解的java NIO的工作原理圖:

Java NIO與阻塞IO的研究

(注:每個線程的處理流程大概都是讀取資料、解碼、計算處理、編碼、發送響應。)

Java NIO的服務端隻需啟動一個專門的線程來處理所有的 IO 事件,這種通信模型是怎麼實作的呢?呵呵,我們一起來探究它的奧秘吧。java NIO采用了雙向通道(channel)進行資料傳輸,而不是單向的流(stream),在通道上可以注冊我們感興趣的事件。一共有以下四種事件:

事件名 對應值
服務端接收用戶端連接配接事件 SelectionKey.OP_ACCEPT(16)
用戶端連接配接服務端事件 SelectionKey.OP_CONNECT(8)
讀事件 SelectionKey.OP_READ(1)
寫事件 SelectionKey.OP_WRITE(4)

服務端和用戶端各自維護一個管理通道的對象,我們稱之為selector,該對象能檢測一個或多個通道 (channel) 上的事件。我們以服務端為例,如果服務端的selector上注冊了讀事件,某時刻用戶端給服務端發送了一些資料,阻塞I/O這時會調用read()方法阻塞地讀取資料,而NIO的服務端會在selector中添加一個讀事件。服務端的處理線程會輪詢地通路selector,如果通路selector時發現有感興趣的事件到達,則處理這些事件,如果沒有感興趣的事件到達,則處理線程會一直阻塞直到感興趣的事件到達為止。下面是我了解的java NIO的通信模型示意圖:

Java NIO與阻塞IO的研究

二.java NIO服務端和用戶端代碼實作 

package test;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
 * Created by Jason on 2016/7/25.
 */
public class NIOServer {

    //通道管理器
    private Selector selector;

    /**
     * 獲得一個ServerSocket通道,并對該通道做一些初始化的工作
     * @param port  綁定的端口号
     * @throws IOException
     */
    public void initServer(int port) throws IOException {
        // 獲得一個ServerSocket通道
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        // 設定通道為非阻塞
        serverChannel.configureBlocking(false);
        // 将該通道對應的ServerSocket綁定到port端口
        serverChannel.socket().bind(new InetSocketAddress(port));
        // 獲得一個通道管理器
        this.selector = Selector.open();
        //将通道管理器和該通道綁定,并為該通道注冊SelectionKey.OP_ACCEPT事件,注冊該事件後,
        //當該事件到達時,selector.select()會傳回,如果該事件沒到達selector.select()會一直阻塞。
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
    }

    /**
     * 采用輪詢的方式監聽selector上是否有需要處理的事件,如果有,則進行處理
     * @throws IOException
     */
    @SuppressWarnings("unchecked")
    public void listen() throws IOException {
        System.out.println("服務端啟動成功!");
        // 輪詢通路selector
        while (true) {
            //當注冊的事件到達時,方法傳回;否則,該方法會一直阻塞
            selector.select();
            // 獲得selector中選中的項的疊代器,選中的項為注冊的事件
            Iterator ite = this.selector.selectedKeys().iterator();
            while (ite.hasNext()) {
                SelectionKey key = (SelectionKey) ite.next();
                // 删除已選的key,以防重複處理
                ite.remove();
                // 用戶端請求連接配接事件
                if (key.isAcceptable()) {
                    ServerSocketChannel server = (ServerSocketChannel) key
                            .channel();
                    // 獲得和用戶端連接配接的通道
                    SocketChannel channel = server.accept();
                    // 設定成非阻塞
                    channel.configureBlocking(false);

                    //在這裡可以給用戶端發送資訊哦
                    channel.write(ByteBuffer.wrap(new String("向用戶端發送了一條資訊").getBytes()));
                    //在和用戶端連接配接成功之後,為了可以接收到用戶端的資訊,需要給通道設定讀的權限。
                    channel.register(this.selector, SelectionKey.OP_READ);

                    // 獲得了可讀的事件
                } else if (key.isReadable()) {
                    read(key);
                }

            }

        }
    }
    /**
     * 處理讀取用戶端發來的資訊 的事件
     * @param key
     * @throws IOException
     */
    public void read(SelectionKey key) throws IOException{
        // 伺服器可讀取消息:得到事件發生的Socket通道
        SocketChannel channel = (SocketChannel) key.channel();
        // 建立讀取的緩沖區
        ByteBuffer buffer = ByteBuffer.allocate(10);
        channel.read(buffer);
        byte[] data = buffer.array();
        String msg = new String(data).trim();
        System.out.println("服務端收到資訊:"+msg);
        ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
        channel.write(outBuffer);// 将消息回送給用戶端
    }

    /**
     * 啟動服務端測試
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
        NIOServer server = new NIOServer();
        server.initServer(8000);
        server.listen();
    }
}
      

用戶端:

package test;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
 * Created by Jason on 2016/7/25.
 */
public class NIOClient {
    //通道管理器
    private Selector selector;

    /**
     * 獲得一個Socket通道,并對該通道做一些初始化的工作
     * @param ip 連接配接的伺服器的ip
     * @param port  連接配接的伺服器的端口号
     * @throws IOException
     */
    public void initClient(String ip,int port) throws IOException {
        // 獲得一個Socket通道
        SocketChannel channel = SocketChannel.open();
        // 設定通道為非阻塞
        channel.configureBlocking(false);
        // 獲得一個通道管理器
        this.selector = Selector.open();

        // 用戶端連接配接伺服器,其實方法執行并沒有實作連接配接,需要在listen()方法中調
        //用channel.finishConnect();才能完成連接配接
        channel.connect(new InetSocketAddress(ip,port));
        //将通道管理器和該通道綁定,并為該通道注冊SelectionKey.OP_CONNECT事件。
        channel.register(selector, SelectionKey.OP_CONNECT);
    }

    /**
     * 采用輪詢的方式監聽selector上是否有需要處理的事件,如果有,則進行處理
     * @throws IOException
     */
    @SuppressWarnings("unchecked")
    public void listen() throws IOException {
        // 輪詢通路selector
        while (true) {
            selector.select();
            // 獲得selector中選中的項的疊代器
            Iterator ite = this.selector.selectedKeys().iterator();
            while (ite.hasNext()) {
                SelectionKey key = (SelectionKey) ite.next();
                // 删除已選的key,以防重複處理
                ite.remove();
                // 連接配接事件發生
                if (key.isConnectable()) {
                    SocketChannel channel = (SocketChannel) key
                            .channel();
                    // 如果正在連接配接,則完成連接配接
                    if(channel.isConnectionPending()){
                        channel.finishConnect();

                    }
                    // 設定成非阻塞
                    channel.configureBlocking(false);

                    //在這裡可以給服務端發送資訊哦
                    channel.write(ByteBuffer.wrap(new String("向服務端發送了一條資訊").getBytes()));
                    //在和服務端連接配接成功之後,為了可以接收到服務端的資訊,需要給通道設定讀的權限。
                    channel.register(this.selector, SelectionKey.OP_READ);

                    // 獲得了可讀的事件
                } else if (key.isReadable()) {
                    read(key);
                }

            }

        }
    }
    /**
     * 處理讀取服務端發來的資訊 的事件
     * @param key
     * @throws IOException
     */
    public void read(SelectionKey key) throws IOException{
        //和服務端的read方法一樣
    }


    /**
     * 啟動用戶端測試
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
        NIOClient client = new NIOClient();
        client.initClient("localhost",8000);
        client.listen();
    }
}
      

以上借鑒分享資料整理:  http://weixiaolu.iteye.com/blog/1479656

繼續閱讀