Java?IO網(wǎng)絡(luò)模型實現(xiàn)解析

更新時間：2023年03月16日 11:46:21 作者：半夏之沫

這篇文章主要為大家介紹了Java?IO網(wǎng)絡(luò)模型實現(xiàn)解析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

本篇文章會對Java中的網(wǎng)絡(luò)IO模型的概念進(jìn)行解釋，并給出具體的Java代碼實現(xiàn)，主要涉及如下部分。

BIO（同步阻塞IO模型）的概念和Java編程實現(xiàn)；
Non Blocking IO（同步非阻塞IO模型）的概念和Java編程實現(xiàn)；
IO多路復(fù)用的概念；
NIO（New IO）的概念和Java編程實現(xiàn)。

在開始本篇文章內(nèi)容之前，有一個簡單的關(guān)于Socket的知識需要說明：在進(jìn)行網(wǎng)絡(luò)通信的時候，需要一對Socket，一個運行于客戶端，一個運行于服務(wù)端，同時服務(wù)端還會有一個服務(wù)端Socket，用于監(jiān)聽客戶端的連接。下圖進(jìn)行一個簡單示意。

那么整個通信流程如下所示。

服務(wù)端運行后，會在服務(wù)端創(chuàng)建listen-socket，listen-socket會綁定服務(wù)端的ip和port，然后服務(wù)端進(jìn)入監(jiān)聽狀態(tài)；
客戶端請求服務(wù)端時，客戶端創(chuàng)建connect-socket，connect-socket描述了其要連接的服務(wù)端的listen-socket，然后connect-socket向listen-socket發(fā)起連接請求；
connect-socket與listen-socket成功連接后（TCP三次握手成功），服務(wù)端會為已連接的客戶端創(chuàng)建一個代表該客戶端的client-socket，用于后續(xù)和客戶端進(jìn)行通信；
客戶端與服務(wù)端通過socket進(jìn)行網(wǎng)絡(luò)IO操作，此時就實現(xiàn)了客戶端和服務(wù)端中的不同進(jìn)程的通信。

需要知道的就是，在客戶端與服務(wù)端通信的過程中，出現(xiàn)了三種socket，分別是。

listen-socket。是服務(wù)端用于監(jiān)聽客戶端建立連接的socket；
connect-socket。是客戶端用于連接服務(wù)端的socket；
client-socket。是服務(wù)端監(jiān)聽到客戶端連接請求后，在服務(wù)端生成的與客戶端連接的socket。

（注：上述中的socket，可以被稱為套接字，也可以被稱為文件描述符。）

正文

一. BIO

BIO，即同步阻塞IO模型。用戶進(jìn)程調(diào)用read時發(fā)起IO操作，此時用戶進(jìn)程由用戶態(tài)轉(zhuǎn)換到內(nèi)核態(tài)，只有在內(nèi)核態(tài)中將IO操作執(zhí)行完后，才會從內(nèi)核態(tài)切換回用戶態(tài)，這期間用戶進(jìn)程會一直阻塞。

BIO示意圖如下。

簡單的BIO的Java編程實現(xiàn)如下。

服務(wù)端實現(xiàn)

public class BioServer {

    public static void main(String[] args) throws IOException {
        // 創(chuàng)建listen-socket
        ServerSocket listenSocket = new ServerSocket(8080);
        // 進(jìn)入監(jiān)聽狀態(tài)，是一個阻塞狀態(tài)
        // 有客戶端連接時從監(jiān)聽狀態(tài)返回
        // 并創(chuàng)建代表這個客戶端的client-socket
        Socket clientSocket = listenSocket.accept();
        // 獲取client-socket輸入流
        BufferedReader bufferedReader = new BufferedReader(
                new InputStreamReader(clientSocket.getInputStream()));
        // 讀取客戶端發(fā)送的數(shù)據(jù)
        // 如果數(shù)據(jù)沒準(zhǔn)備好，會進(jìn)入阻塞狀態(tài)
        String data = bufferedReader.readLine();
        System.out.println(data);

        // 獲取client-socket輸出流
        BufferedWriter bufferedWriter = new BufferedWriter(
                new OutputStreamWriter(clientSocket.getOutputStream()));
        // 服務(wù)端向客戶端發(fā)送數(shù)據(jù)
        bufferedWriter.write("來自服務(wù)端的返回數(shù)據(jù)\n");
        // 刷新流
        bufferedWriter.flush();
    }

}

客戶端實現(xiàn)

public class BioClient {

    public static final String SERVER_IP = "127.0.0.1";
    public static final int SERVER_PORT = 8080;

    public static void main(String[] args) throws IOException {
        // 客戶端創(chuàng)建connect-socket
        Socket connectSocket = new Socket(SERVER_IP, SERVER_PORT);
        // 獲取connect-socket輸出流
        BufferedWriter bufferedWriter = new BufferedWriter(
                new OutputStreamWriter(connectSocket.getOutputStream()));
        // 客戶端向服務(wù)端發(fā)送數(shù)據(jù)
        bufferedWriter.write("來自客戶端的請求數(shù)據(jù)\n");
        // 刷新流
        bufferedWriter.flush();

        // 獲取connect-socket輸入流
        BufferedReader bufferedReader = new BufferedReader(
                new InputStreamReader(connectSocket.getInputStream()));
        // 讀取服務(wù)端發(fā)送的數(shù)據(jù)
        String returnData = bufferedReader.readLine();
        System.out.println(returnData);
    }

}

BIO的問題就在于服務(wù)端在accept時是阻塞的，并且在主線程中，一次只能accept一個Socket，accept到Socket后，讀取客戶端數(shù)據(jù)時又是阻塞的。

二. Non Blocking IO

Non Blocking IO，即同步非阻塞IO。是用戶進(jìn)程調(diào)用read時，用戶進(jìn)程由用戶態(tài)轉(zhuǎn)換到內(nèi)核態(tài)后，此時如果沒有系統(tǒng)資源數(shù)據(jù)能夠被讀取到內(nèi)核緩沖區(qū)中，返回read失敗，并從內(nèi)核態(tài)切換回用戶態(tài)。也就是用戶進(jìn)程發(fā)起IO操作后會立即得到一個操作結(jié)果。

Non Blocking IO示意圖如下所示。

簡單的Non Blocking IO的Java編程實現(xiàn)如下。

public class NonbioServer {

    public static final List&lt;SocketChannel&gt; clientSocketChannels = new ArrayList&lt;&gt;();

    public static void main(String[] args) throws Exception {
        // 客戶端創(chuàng)建listen-socket管道
        // 管道支持非阻塞模式和同時讀寫
        ServerSocketChannel listenSocketChannel = ServerSocketChannel.open();
        // 設(shè)置為非阻塞模式
        listenSocketChannel.configureBlocking(false);
        // 綁定監(jiān)聽的端口號
        listenSocketChannel.socket().bind(new InetSocketAddress(8080));
        // 在子線程中遍歷clientSocketChannels并讀取客戶端數(shù)據(jù)
        handleSocketChannels();

        while (true) {
            // 非阻塞方式監(jiān)聽客戶端連接
            // 如果無客戶端連接則返回空
            // 有客戶端連接則創(chuàng)建代表這個客戶端的client-socket管道
            SocketChannel clientSocketChannel = listenSocketChannel.accept();
            if (clientSocketChannel != null) {
                // 設(shè)置為非阻塞模式
                clientSocketChannel.configureBlocking(false);
                // 添加到clientSocketChannels中
                // 用于子線程遍歷并讀取客戶端數(shù)據(jù)
                clientSocketChannels.add(clientSocketChannel);
            } else {
                LockSupport.parkNanos(1000 * 1000 * 1000);
            }
        }
    }

    public static void handleSocketChannels() {
        new Thread(() -&gt; {
            while (true) {
                // 遍歷每一個client-socket管道
                Iterator&lt;SocketChannel&gt; iterator = clientSocketChannels.iterator();
                while (iterator.hasNext()) {
                    SocketChannel clientSocketChannel = iterator.next();
                    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                    int read = 0;
                    try {
                        // 將客戶端發(fā)送的數(shù)據(jù)讀取到ByteBuffer中
                        // 這一步的操作也是非阻塞的
                        read = clientSocketChannel.read(byteBuffer);
                    } catch (IOException e) {
                        // 移除發(fā)生異常的client-socket管道
                        iterator.remove();
                        e.printStackTrace();
                    }
                    if (read == 0) {
                        System.out.println("客戶端數(shù)據(jù)未就緒");
                    } else {
                        System.out.println("客戶端數(shù)據(jù)為：" + new String(byteBuffer.array()));
                    }
                }
                LockSupport.parkNanos(1000 * 1000 * 1000);
            }
        }).start();
    }

}

上述是Non Blocking IO的一個簡單服務(wù)端的實現(xiàn)，相較于BIO，服務(wù)端在accept時是非阻塞的，在讀取客戶端數(shù)據(jù)時也是非阻塞的，但是還是存在如下問題。

一次只能accept一個Socket；
需要在用戶進(jìn)程中遍歷所有的SocketChannel并調(diào)用read() 方法獲取客戶端數(shù)據(jù)，此時如果客戶端數(shù)據(jù)未準(zhǔn)備就緒，那么這一次的read() 操作的開銷就是浪費的。

三. IO多路復(fù)用

在上述的BIO和Non Blocking IO中，一次系統(tǒng)調(diào)用，只會獲取一個IO的狀態(tài)，而如果采取IO多路復(fù)用機制，則可以一次系統(tǒng)調(diào)用獲取多個IO的狀態(tài)。

也就是獲取多個IO的狀態(tài)可以復(fù)用一次系統(tǒng)調(diào)用。

最簡單的IO多路復(fù)用方式是基于select模型實現(xiàn)，步驟如下。

在用戶進(jìn)程中將需要監(jiān)控的IO文件描述符（Socket）注冊到IO多路復(fù)用器中；
執(zhí)行select操作，此時用戶進(jìn)程由用戶態(tài)轉(zhuǎn)換到內(nèi)核態(tài)（一次系統(tǒng)調(diào)用），然后在內(nèi)核態(tài)中會輪詢注冊到IO多路復(fù)用器中的IO是否準(zhǔn)備就緒，并得到所有準(zhǔn)備就緒的IO的文件描述符列表，最后返回這些文件描述符列表；
用戶進(jìn)程在select操作返回前會一直阻塞，直至select操作返回，此時用戶進(jìn)程就獲得了所有就緒的IO的文件描述符列表；
用戶進(jìn)程獲得了就緒的IO的文件描述符列表后，就可以對這些IO進(jìn)行相應(yīng)的操作了。

換言之，IO多路復(fù)用中，只需要一次系統(tǒng)調(diào)用，IO多路復(fù)用器就可以告訴用戶進(jìn)程，哪些IO已經(jīng)準(zhǔn)備就緒可以進(jìn)行操作了，而如果不采用IO多路復(fù)用，則需要用戶進(jìn)程自己遍歷每個IO并調(diào)用accept() 或者read() 方法去判斷，且一次accept() 或者read() 方法調(diào)用只能判斷一個IO。

四. NIO

NIO，即New IO。關(guān)于NIO，有如下三大組件。

channel（管道）。介于buffer（字節(jié)緩沖區(qū)）和Socket（套接字）之間，用于數(shù)據(jù)的讀寫操作；
buffer（字節(jié)緩沖區(qū)）。是用戶程序和channel（管道）之間進(jìn)行讀寫數(shù)據(jù)的中間區(qū)域；
selector（IO多路復(fù)用器）。服務(wù)端的listen-socket和client-socket，客戶端的connect-socket，都可以注冊在selector上，注冊的時候還需要指定監(jiān)聽的事件，比如為listen-socket指定監(jiān)聽的事件為ACCEPT事件，該事件發(fā)生則表示客戶端建立了連接，還比如為client-socket指定監(jiān)聽的事件為READ事件，該事件發(fā)生則表示客戶端發(fā)送的數(shù)據(jù)已經(jīng)可讀。

NIO的代碼實現(xiàn)如下所示。

服務(wù)端實現(xiàn)

public class NioServer {

    private static Selector selector;

    public static void main(String[] args) {

        try {
            // 開啟并得到多路復(fù)用器
            selector = Selector.open();
            // 服務(wù)端創(chuàng)建listen-socket管道
            ServerSocketChannel listenSocketChannel = ServerSocketChannel.open();
            // 設(shè)置為非阻塞模式
            listenSocketChannel.configureBlocking(false);
            // 為管道綁定端口
            listenSocketChannel.socket().bind(new InetSocketAddress(8080));
            // 將listen-socket管道注冊到多路復(fù)用器上，并指定監(jiān)聽ACCEPT事件
            listenSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

            while (true) {
                // 獲取發(fā)生的事件，這個操作是阻塞的
                selector.select();
                // 拿到有事件發(fā)生的SelectionKey集合
                // SelectionKey表示管道與多路復(fù)用器的綁定關(guān)系
                Set&lt;SelectionKey&gt; selectionKeys = selector.selectedKeys();
                // 遍歷每個發(fā)生的事件，然后判斷事件類型
                // 根據(jù)事件類型，進(jìn)行不同的處理
                Iterator&lt;SelectionKey&gt; iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();
                    if (selectionKey.isAcceptable()) {
                        // 處理客戶端連接事件
                        handlerAccept(selectionKey);
                    } else if (selectionKey.isReadable()) {
                        // 處理客戶端數(shù)據(jù)可讀事件
                        handlerRead(selectionKey);
                    }
                }
                LockSupport.parkNanos(1000 * 1000 * 1000);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void handlerAccept(SelectionKey selectionKey) {
        // 從事件中獲取到listen-socket管道
        ServerSocketChannel listenSocketChannel = (ServerSocketChannel) selectionKey.channel();
        try {
            // 為連接的客戶端創(chuàng)建client-socket管道
            SocketChannel clientSocketChannel = listenSocketChannel.accept();
            // 設(shè)置為非阻塞模式
            clientSocketChannel.configureBlocking(false);
            // 將client-socket管道注冊到多路復(fù)用器上，并指定監(jiān)聽READ事件
            clientSocketChannel.register(selector, SelectionKey.OP_READ);
            // 給客戶端發(fā)送數(shù)據(jù)
            clientSocketChannel.write(ByteBuffer.wrap("連接已建立\n".getBytes()));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void handlerRead(SelectionKey selectionKey) {
        // 從事件中獲取到client-socket管道
        SocketChannel clientSocketChannel = (SocketChannel) selectionKey.channel();
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        try {
            // 讀取客戶端數(shù)據(jù)
            int read = clientSocketChannel.read(byteBuffer);
            if (read &lt;= 0) {
                // 關(guān)閉管道
                clientSocketChannel.close();
                // 從多路復(fù)用器移除綁定關(guān)系
                selectionKey.cancel();
            } else {
                System.out.println(new String(byteBuffer.array()));
            }
        } catch (IOException e1) {
            try {
                // 關(guān)閉管道
                clientSocketChannel.close();
            } catch (IOException e2) {
                e2.printStackTrace();
            }
            // 從多路復(fù)用器移除綁定關(guān)系
            selectionKey.cancel();
            e1.printStackTrace();
        }
    }

}

客戶端實現(xiàn)

public class NioClient {

    private static Selector selector;

    public static final String SERVER_IP = "127.0.0.1";
    public static final int SERVER_PORT = 8080;

    public static void main(String[] args) {
        try {
            // 開啟并得到多路復(fù)用器
            selector = Selector.open();
            // 創(chuàng)建connect-socket管道
            SocketChannel connectSocketChannel = SocketChannel.open();
            // 設(shè)置為非阻塞模式
            connectSocketChannel.configureBlocking(false);
            // 設(shè)置服務(wù)端IP和端口
            connectSocketChannel.connect(new InetSocketAddress(SERVER_IP, SERVER_PORT));
            // 將connect-socket管道注冊到多路復(fù)用器上，并指定監(jiān)聽CONNECT事件
            connectSocketChannel.register(selector, SelectionKey.OP_CONNECT);

            while (true) {
                // 獲取發(fā)生的事件，這個操作是阻塞的
                selector.select();
                // 拿到有事件發(fā)生的SelectionKey集合
                // SelectionKey表示管道與多路復(fù)用器的綁定關(guān)系
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                // 遍歷每個發(fā)生的事件，然后判斷事件類型
                // 根據(jù)事件類型，進(jìn)行不同的處理
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();
                    if (selectionKey.isConnectable()) {
                        // 處理連接建立事件
                        handlerConnect(selectionKey);
                    } else if (selectionKey.isReadable()) {
                        // 處理服務(wù)端數(shù)據(jù)可讀事件
                        handlerRead(selectionKey);
                    }
                }
                LockSupport.parkNanos(1000 * 1000 * 1000);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void handlerConnect(SelectionKey selectionKey) throws IOException {
        // 拿到connect-socket管道
        SocketChannel connectSocketChannel = (SocketChannel) selectionKey.channel();
        if (connectSocketChannel.isConnectionPending()) {
            connectSocketChannel.finishConnect();
        }
        // 設(shè)置為非阻塞模式
        connectSocketChannel.configureBlocking(false);
        // 將connect-socket管道注冊到多路復(fù)用器上，并指定監(jiān)聽READ事件
        connectSocketChannel.register(selector, SelectionKey.OP_READ);
        // 向服務(wù)端發(fā)送數(shù)據(jù)
        connectSocketChannel.write(ByteBuffer.wrap("客戶端發(fā)送的數(shù)據(jù)\n".getBytes()));
    }

    private static void handlerRead(SelectionKey selectionKey) {
        // 拿到connect-socket管道
        SocketChannel connectSocketChannel = (SocketChannel) selectionKey.channel();
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        try {
            // 讀取服務(wù)端數(shù)據(jù)
            int read = connectSocketChannel.read(byteBuffer);
            if (read <= 0) {
                // 關(guān)閉管道
                connectSocketChannel.close();
                // 從多路復(fù)用器移除綁定關(guān)系
                selectionKey.cancel();
            } else {
                System.out.println(new String(byteBuffer.array()));
            }
        } catch (IOException e1) {
            try {
                // 關(guān)閉管道
                connectSocketChannel.close();
            } catch (IOException e2) {
                e2.printStackTrace();
            }
            // 從多路復(fù)用器移除綁定關(guān)系
            selectionKey.cancel();
            e1.printStackTrace();
        }
    }

}

總結(jié)

本篇文章中所討論的IO模型，都是同步IO模型，而何謂同步異步，何謂阻塞非阻塞，可以總結(jié)如下。

同步和異步。同步IO表示需要在用戶進(jìn)程中主動的去詢問操作系統(tǒng)數(shù)據(jù)是否準(zhǔn)備好，如果沒有準(zhǔn)備好，還需要持續(xù)的詢問，直到數(shù)據(jù)準(zhǔn)備好為止；而異步IO則是在用戶進(jìn)程中只需要詢問操作系統(tǒng)一次，后續(xù)數(shù)據(jù)準(zhǔn)備好后操作系統(tǒng)會主動的將數(shù)據(jù)給到用戶進(jìn)程；
阻塞和非阻塞。阻塞IO就是發(fā)起一次系統(tǒng)調(diào)用后，會一直阻塞直到有結(jié)果返回；而非阻塞IO就是發(fā)起一次系統(tǒng)調(diào)用后，會立即得到一個返回結(jié)果。

實際上在Java中是有對異步IO（AIO）做支持，但是AIO依賴操作系統(tǒng)的底層實現(xiàn)，而目前Linux對AIO的支持不成熟，所以AIO的使用并不多，像主流的網(wǎng)絡(luò)應(yīng)用框架Netty也都沒有使用到AIO。

以上就是Java IO網(wǎng)絡(luò)模型實現(xiàn)解析的詳細(xì)內(nèi)容，更多關(guān)于Java IO網(wǎng)絡(luò)模型的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: