大家好，我是老三，上一節(jié)我們討論了Linux的五種IO模型，接下來(lái)，我們從Java語(yǔ)言層面，來(lái)看看對(duì)IO的實(shí)現(xiàn)。

在Java中，一共有三種IO模型，分別是阻塞IO(BIO)、非阻塞IO(NIO)和異步IO(AIO)。

(相關(guān)資料圖)

Linux五種IO模型和Java三種IO模型

Java BIO

Java BIO就是Java的傳統(tǒng)IO模型，對(duì)應(yīng)了操作系統(tǒng)IO模型里的阻塞IO。

Java BIO相關(guān)的實(shí)現(xiàn)都位于java.io包下，其通信原理是客戶端、服務(wù)端之間通過(guò)Socket套接字建立管道連接，然后從管道中獲取對(duì)應(yīng)的輸入/輸出流，最后利用輸入/輸出流對(duì)象實(shí)現(xiàn)發(fā)送/接收信息。

我們來(lái)看個(gè)Demo：

BioServer：

/** * @Author 三分惡 * @Date 2023/4/30 * @Description BIO服務(wù)端 */public class BioServer {    public static void main(String[] args) throws IOException {        //定義一個(gè)ServerSocket服務(wù)端對(duì)象，并為其綁定端口號(hào)        ServerSocket server = new ServerSocket(8888);        System.out.println("===========BIO服務(wù)端啟動(dòng)================");        //對(duì)BIO來(lái)講，每個(gè)Socket都需要一個(gè)Thread        while (true) {            //監(jiān)聽(tīng)客戶端Socket連接            Socket socket = server.accept();            new BioServerThread(socket).start();        }    }    /**     * BIO Server線程     */    static class BioServerThread extends Thread{        //socket連接        private Socket socket;        public BioServerThread(Socket socket){            this.socket=socket;        }        @Override        public void run() {            try {                //從socket中獲取輸入流                InputStream inputStream=socket.getInputStream();                //轉(zhuǎn)換為                BufferedReader bufferedReader=new BufferedReader(new InputStreamReader(inputStream));                String msg;                //從Buffer中讀取信息，如果讀取到信息則輸出                while((msg=bufferedReader.readLine())!=null){                    System.out.println("收到客戶端消息："+msg);                }                //從socket中獲取輸出流                OutputStream outputStream=socket.getOutputStream();                PrintStream printStream=new PrintStream(outputStream);                //通過(guò)輸出流對(duì)象向客戶端傳遞信息                printStream.println("你好，吊毛！");                //清空輸出流                printStream.flush();                //關(guān)閉socket                socket.shutdownOutput();            } catch (IOException e) {                e.printStackTrace();            }        }    }}

BioClient

/** * @Author 三分惡 * @Date 2023/4/30 * @Description BIO客戶端 */public class BioClient {    public static void main(String[] args) throws IOException {        List names= Arrays.asList("帥哥","靚仔","坤坤");        //通過(guò)循環(huán)創(chuàng)建多個(gè)多個(gè)client        for (String name:names){            //創(chuàng)建socket并根據(jù)IP地址與端口連接服務(wù)端            Socket socket=new Socket("127.0.0.1",8888);            System.out.println("===========BIO客戶端啟動(dòng)================");            //從socket中獲取字節(jié)輸出流            OutputStream outputStream=socket.getOutputStream();            //通過(guò)輸出流向服務(wù)端傳遞信息            String hello="你好，"+name+"!";            outputStream.write(hello.getBytes());            //清空流，關(guān)閉socket輸出            outputStream.flush();            socket.shutdownOutput();            //從socket中獲取字節(jié)輸入流            InputStream inputStream=socket.getInputStream();            BufferedReader bufferedReader=new BufferedReader(new InputStreamReader(inputStream));            //讀取服務(wù)端消息            String msg;            while((msg=bufferedReader.readLine())!=null){                System.out.println("收到服務(wù)端消息："+msg);            }            inputStream.close();            outputStream.close();            socket.close();        }    }}

先啟動(dòng)BioServer，再啟動(dòng)BioClient，運(yùn)行結(jié)果

===========BIO服務(wù)端啟動(dòng)================收到客戶端消息：你好，帥哥!收到客戶端消息：你好，靚仔!收到客戶端消息：你好，坤坤!

===========BIO客戶端啟動(dòng)================收到服務(wù)端消息：你好，吊毛！===========BIO客戶端啟動(dòng)================收到服務(wù)端消息：你好，吊毛！===========BIO客戶端啟動(dòng)================收到服務(wù)端消息：你好，吊毛！

在上述Java-BIO的通信過(guò)程中，如果客戶端一直沒(méi)有發(fā)送消息過(guò)來(lái)，服務(wù)端則會(huì)一直等待下去，從而服務(wù)端陷入阻塞狀態(tài)。同理，由于客戶端也一直在等待服務(wù)端的消息，如果服務(wù)端一直未響應(yīng)消息回來(lái)，客戶端也會(huì)陷入阻塞狀態(tài)。

在BioServer定義了一個(gè)類BioServerThread，繼承了Thread類，run方法里主要是通過(guò)socket和流來(lái)讀取客戶端的消息，以及發(fā)送消息給客戶端，每處理一個(gè)客戶端的Socket連接，就得新建一個(gè)線程。

同時(shí)，IO讀寫操作也是阻塞的，如果客戶端一直沒(méi)有發(fā)送消息過(guò)來(lái)，線程就會(huì)進(jìn)入阻塞狀態(tài)，一直等待下去。

在BioClient里，循環(huán)創(chuàng)建Socket，向服務(wù)端收發(fā)消息，客戶端的讀寫也是阻塞的。

在這個(gè)Demo里就體現(xiàn)了BIO的兩個(gè)特點(diǎn)：

一個(gè)客戶端連接對(duì)應(yīng)一個(gè)處理線程讀寫操作都是阻塞的

Java BIO

毫無(wú)疑問(wèn)，不管是創(chuàng)建太多線程，還是阻塞讀寫，都會(huì)浪費(fèi)服務(wù)器的資源。

Java NIO

那么我們就進(jìn)入Java的下一種IO模型——Java NIO，它對(duì)應(yīng)操作系統(tǒng)IO模型中的多路復(fù)用IO，底層采用了epoll實(shí)現(xiàn)。

Java-NIO則是JDK1.4中新引入的API，它在BIO功能的基礎(chǔ)上實(shí)現(xiàn)了非阻塞式的特性，其所有實(shí)現(xiàn)都位于java.nio包下。NIO是一種基于通道、面向緩沖區(qū)的IO操作，相較BIO而言，它能夠更為高效的對(duì)數(shù)據(jù)進(jìn)行讀寫操作，同時(shí)與原先的BIO使用方式也大有不同。

我們還是先來(lái)看個(gè)Demo：

NioServer

/** * @Author 三分惡 * @Date 2023/4/30 * @Description NIO服務(wù)端 */public class NioServer {    public static void main(String[] args) throws IOException {        //創(chuàng)建一個(gè)選擇器selector        Selector selector= Selector.open();        //創(chuàng)建serverSocketChannel        ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();        //綁定端口        serverSocketChannel.socket().bind(new InetSocketAddress(8888));        //必須得設(shè)置成非阻塞模式        serverSocketChannel.configureBlocking(false);        //將channel注冊(cè)到selector并設(shè)置監(jiān)聽(tīng)事件為ACCEPT        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);        System.out.println("===========NIO服務(wù)端啟動(dòng)============");        while(true){            //超時(shí)等待            if(selector.select(1000)==0){                System.out.println("===========NIO服務(wù)端超時(shí)等待============");                continue;            }            // 有客戶端請(qǐng)求被輪詢監(jiān)聽(tīng)到，獲取返回的SelectionKey集合            Iterator iterator=selector.selectedKeys().iterator();            //迭代器遍歷SelectionKey集合            while (iterator.hasNext()){                SelectionKey key=iterator.next();                // 判斷是否為ACCEPT事件                if (key.isAcceptable()){                    // 處理接收請(qǐng)求事件                    SocketChannel socketChannel=((ServerSocketChannel) key.channel()).accept();                    //非阻塞模式                    socketChannel.configureBlocking(false);                    // 注冊(cè)到Selector并設(shè)置監(jiān)聽(tīng)事件為READ                    socketChannel.register(selector,SelectionKey.OP_READ, ByteBuffer.allocate(1024));                    System.out.println("成功連接客戶端");                }                //判斷是否為READ事件                if (key.isReadable()){                    SocketChannel socketChannel = (SocketChannel) key.channel();                    try {                        // 獲取以前設(shè)置的附件對(duì)象，如果沒(méi)有則新建一個(gè)                        ByteBuffer buffer = (ByteBuffer) key.attachment();                        if (buffer == null) {                            buffer = ByteBuffer.allocate(1024);                            key.attach(buffer);                        }                        // 清空緩沖區(qū)                        buffer.clear();                        // 將通道中的數(shù)據(jù)讀到緩沖區(qū)                        int len = socketChannel.read(buffer);                        if (len > 0) {                            buffer.flip();                            String message = new String(buffer.array(), 0, len);                            System.out.println("收到客戶端消息：" + message);                        } else if (len < 0) {                            // 接收到-1，表示連接已關(guān)閉                            key.cancel();                            socketChannel.close();                            continue;                        }                        // 注冊(cè)寫事件，下次向客戶端發(fā)送消息                        socketChannel.register(selector, SelectionKey.OP_WRITE, buffer);                    } catch (IOException e) {                        // 取消SelectionKey并關(guān)閉對(duì)應(yīng)的SocketChannel                        key.cancel();                        socketChannel.close();                    }                }                //判斷是否為WRITE事件                if (key.isWritable()){                    SocketChannel socketChannel = (SocketChannel) key.channel();                    //獲取buffer                    ByteBuffer buffer = (ByteBuffer) key.attachment();                    String hello = "你好，坤坤！";                    //清空buffer                    buffer.clear();                    //buffer中寫入消息                    buffer.put(hello.getBytes());                    buffer.flip();                    //向channel中寫入消息                    socketChannel.write(buffer);                    buffer.clear();                    System.out.println("向客戶端發(fā)送消息：" + hello);                    // 設(shè)置下次讀寫操作，向 Selector 進(jìn)行注冊(cè)                    socketChannel.register(selector, SelectionKey.OP_READ, buffer);                }                // 移除本次處理的SelectionKey,防止重復(fù)處理                iterator.remove();            }        }    }}

NioClient

public class NioClient {    public static void main(String[] args) throws IOException {        // 創(chuàng)建SocketChannel并指定ip地址和端口號(hào)        SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8888));        System.out.println("==============NIO客戶端啟動(dòng)================");        // 非阻塞模式        socketChannel.configureBlocking(false);        String hello="你好，靚仔！";        ByteBuffer buffer = ByteBuffer.wrap(hello.getBytes());        // 向通道中寫入數(shù)據(jù)        socketChannel.write(buffer);        System.out.println("發(fā)送消息：" + hello);        buffer.clear();        // 將channel注冊(cè)到Selector并監(jiān)聽(tīng)READ事件        socketChannel.register(Selector.open(), SelectionKey.OP_READ, buffer);        while (true) {            // 讀取服務(wù)端數(shù)據(jù)            if (socketChannel.read(buffer) > 0) {                buffer.flip();                String msg = new String(buffer.array(), 0, buffer.limit());                System.out.println("收到服務(wù)端消息：" + msg);                break;            }        }        // 關(guān)閉輸入流        socketChannel.shutdownInput();        // 關(guān)閉SocketChannel連接        socketChannel.close();    }}

先運(yùn)行NioServer，再運(yùn)行NioClient，運(yùn)行結(jié)果：

===========NIO服務(wù)端啟動(dòng)=======================NIO服務(wù)端超時(shí)等待=======================NIO服務(wù)端超時(shí)等待============成功連接客戶端收到客戶端消息：你好，靚仔！向客戶端發(fā)送消息：你好，坤坤！

==============NIO客戶端啟動(dòng)================發(fā)送消息：你好，靚仔！收到服務(wù)端消息：你好，坤坤！

我們?cè)谶@個(gè)案例里實(shí)現(xiàn)了一個(gè)比較簡(jiǎn)單的Java NIO 客戶端服務(wù)端通信，里面有兩個(gè)小的點(diǎn)需要注意，注冊(cè)到選擇器上的通道都必須要為非阻塞模型，同時(shí)通過(guò)緩沖區(qū)傳輸數(shù)據(jù)時(shí)，必須要調(diào)用flip()方法切換為讀取模式。

代碼流程示意圖

Java-NIO中有三個(gè)核心概念：**Buffer（緩沖區(qū)）、Channel（通道）、Selector（選擇器）**。

Java NIO

每個(gè)客戶端連連接本質(zhì)上對(duì)應(yīng)著一個(gè)Channel通道，每個(gè)通道都有自己的Buffer緩沖區(qū)來(lái)進(jìn)行讀寫，這些Channel被Selector選擇器管理調(diào)度Selector負(fù)責(zé)輪詢所有已注冊(cè)的Channel，監(jiān)聽(tīng)到有事件發(fā)生，才提交給服務(wù)端線程處理，服務(wù)端線程不需要做任何阻塞等待，直接在Buffer里處理Channel事件的數(shù)據(jù)即可，處理完馬上結(jié)束，或返回線程池供其他客戶端事件繼續(xù)使用。通過(guò)Selector，服務(wù)端的一個(gè)Thread就可以處理多個(gè)客戶端的請(qǐng)求Buffer（緩沖區(qū)）就是飯店用來(lái)存放食材的儲(chǔ)藏室，當(dāng)服務(wù)員點(diǎn)餐時(shí)，需要從儲(chǔ)藏室中取出食材進(jìn)行制作。Channel（通道）是用于傳輸數(shù)據(jù)的車道，就像飯店里的上菜窗口，可以快速把點(diǎn)好的菜品送到客人的桌上。Selector（選擇器）就是大堂經(jīng)理，負(fù)責(zé)協(xié)調(diào)服務(wù)員、廚師和客人的配合和溝通，以保證整個(gè)就餐過(guò)程的效率和順暢。Java AIO

Java-AIO也被成為NIO2，它是在NIO的基礎(chǔ)上，引入了新的異步通道的概念，并提供了異步文件通道和異步套接字的實(shí)現(xiàn)。

異步通道的實(shí)現(xiàn)體系

它們的主要區(qū)別就在于這個(gè)異步通道，見(jiàn)名知意：使用異步通道去進(jìn)行IO操作時(shí)，所有操作都為異步非阻塞的，當(dāng)調(diào)用read()/write()/accept()/connect()方法時(shí)，本質(zhì)上都會(huì)交由操作系統(tǒng)去完成，比如要接收一個(gè)客戶端的數(shù)據(jù)時(shí)，操作系統(tǒng)會(huì)先將通道中可讀的數(shù)據(jù)先傳入read()回調(diào)方法指定的緩沖區(qū)中，然后再主動(dòng)通知Java程序去處理。

我們還是先來(lái)看個(gè)Demo：

AioServer

/** * @Author 三分惡 * @Date 2023/5/1 * @Description AIO服務(wù)端 */public class AioServer {    public static void main(String[] args) throws Exception {        // 創(chuàng)建異步通道組，處理IO事件        AsynchronousChannelGroup group = AsynchronousChannelGroup.withFixedThreadPool(10, Executors.defaultThreadFactory());        //創(chuàng)建異步服務(wù)器Socket通道，并綁定端口        AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open(group).bind(new InetSocketAddress(8888));        System.out.println("=============AIO服務(wù)端啟動(dòng)=========");        // 異步等待接收客戶端連接        server.accept(null, new CompletionHandler() {            // 創(chuàng)建ByteBuffer            final ByteBuffer buffer = ByteBuffer.allocate(1024);            @Override            public void completed(AsynchronousSocketChannel channel, Object attachment) {                System.out.println("客戶端連接成功");                try {                    buffer.clear();                    // 異步讀取客戶端發(fā)送的消息                    channel.read(buffer, null, new CompletionHandler() {                        @Override                        public void completed(Integer len, Object attachment) {                            buffer.flip();                            String message = new String(buffer.array(), 0, len);                            System.out.println("收到客戶端消息：" + message);                            // 異步發(fā)送消息給客戶端                            channel.write(ByteBuffer.wrap(("你好，阿坤！").getBytes()), null, new CompletionHandler() {                                @Override                                public void completed(Integer result, Object attachment) {                                    // 關(guān)閉輸出流                                    try {                                        channel.shutdownOutput();                                    } catch (IOException e) {                                        e.printStackTrace();                                    }                                }                                @Override                                public void failed(Throwable exc, Object attachment) {                                    exc.printStackTrace();                                    try {                                        channel.close();                                    } catch (IOException e) {                                        e.printStackTrace();                                    }                                }                            });                        }                        @Override                        public void failed(Throwable exc, Object attachment) {                            exc.printStackTrace();                            try {                                channel.close();                            } catch (IOException e) {                                e.printStackTrace();                            }                        }                    });                } catch (Exception e) {                    e.printStackTrace();                }                // 繼續(xù)異步等待接收客戶端連接                server.accept(null, this);            }            @Override            public void failed(Throwable exc, Object attachment) {                exc.printStackTrace();                // 繼續(xù)異步等待接收客戶端連接                server.accept(null, this);            }        });        // 等待所有連接都處理完畢        group.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);    }}

AioClient

/** * @Author 三分惡 * @Date 2023/5/1 * @Description AIO客戶端 */public class AioClient {    public static void main(String[] args) throws Exception {        // 創(chuàng)建異步Socket通道        AsynchronousSocketChannel client = AsynchronousSocketChannel.open();        // 異步連接服務(wù)器        client.connect(new InetSocketAddress("127.0.0.1", 8888), null, new CompletionHandler() {            // 創(chuàng)建ByteBuffer            final ByteBuffer buffer = ByteBuffer.wrap(("你好，靚仔！").getBytes());            @Override            public void completed(Void result, Object attachment) {                // 異步發(fā)送消息給服務(wù)器                client.write(buffer, null, new CompletionHandler() {                    // 創(chuàng)建ByteBuffer                    final ByteBuffer readBuffer = ByteBuffer.allocate(1024);                    @Override                    public void completed(Integer result, Object attachment) {                        readBuffer.clear();                        // 異步讀取服務(wù)器發(fā)送的消息                        client.read(readBuffer, null, new CompletionHandler() {                            @Override                            public void completed(Integer result, Object attachment) {                                readBuffer.flip();                                String msg = new String(readBuffer.array(), 0, result);                                System.out.println("收到服務(wù)端消息：" + msg);                            }                            @Override                            public void failed(Throwable exc, Object attachment) {                                exc.printStackTrace();                                try {                                    client.close();                                } catch (IOException e) {                                    e.printStackTrace();                                }                            }                        });                    }                    @Override                    public void failed(Throwable exc, Object attachment) {                        exc.printStackTrace();                        try {                            client.close();                        } catch (IOException e) {                            e.printStackTrace();                        }                    }                });            }            @Override            public void failed(Throwable exc, Object attachment) {                exc.printStackTrace();                try {                    client.close();                } catch (IOException e) {                    e.printStackTrace();                }            }        });        // 等待連接處理完畢        Thread.sleep(1000);        // 關(guān)閉輸入流和Socket通道        client.shutdownInput();        client.close();    }}

看下運(yùn)行結(jié)果

=============AIO服務(wù)端啟動(dòng)=========客戶端連接成功收到客戶端消息：你好，靚仔！

收到服務(wù)端消息：你好，阿坤！

可以看到，所有的操作都是異步進(jìn)行，通過(guò)completed接收異步回調(diào)，通過(guò)failed接收錯(cuò)誤回調(diào)。

而且我們發(fā)現(xiàn)，相較于之前的NIO而言，AIO其中少了Selector選擇器這個(gè)核心組件，選擇器在NIO中充當(dāng)了協(xié)調(diào)者的角色。

但在Java-AIO中，類似的角色直接由操作系統(tǒng)擔(dān)當(dāng)，而且不是采用輪詢的方式監(jiān)聽(tīng)I(yíng)O事件，而是采用一種類似于“訂閱-通知”的模式。

Java AIO簡(jiǎn)圖

在AIO中，所有創(chuàng)建的通道都會(huì)直接在OS上注冊(cè)監(jiān)聽(tīng)，當(dāng)出現(xiàn)IO請(qǐng)求時(shí)，會(huì)先由操作系統(tǒng)接收、準(zhǔn)備、拷貝好數(shù)據(jù)，然后再通知監(jiān)聽(tīng)對(duì)應(yīng)通道的程序處理數(shù)據(jù)。

Java-AIO這種異步非阻塞式IO也是由操作系統(tǒng)進(jìn)行支持的，在Windows系統(tǒng)中提供了一種異步IO技術(shù)：IOCP(I/O Completion Port，所以Windows下的Java-AIO則是依賴于這種機(jī)制實(shí)現(xiàn)。不過(guò)在Linux系統(tǒng)中由于沒(méi)有這種異步IO技術(shù)，所以Java-AIO在Linux環(huán)境中使用的還是epoll這種多路復(fù)用技術(shù)進(jìn)行模擬實(shí)現(xiàn)的。

因?yàn)長(zhǎng)inux的異步IO技術(shù)實(shí)際上不太成熟，所以Java-AIO的實(shí)際應(yīng)用并不是太多，比如大名鼎鼎的網(wǎng)絡(luò)通信框架Netty就沒(méi)有采用Java-AIO，而是使用Java-NIO，在代碼層面，自行實(shí)現(xiàn)異步。

小結(jié)

那么這期我們就快速過(guò)了一下Java的三種IO機(jī)制，它們的特點(diǎn)，我們直接看下圖：

Java三種IO模型

我們也發(fā)現(xiàn)，雖然Java-NIO、Java-AIO，在性能上比Java-BIO要強(qiáng)很多，但是可以看到，寫法上一個(gè)比一個(gè)難搞，不過(guò)好在基本也沒(méi)人直接用Java-NIO、Java-AIO，如果要進(jìn)行網(wǎng)絡(luò)通信，一般都會(huì)采用Netty，它對(duì)原生的Java-NIO進(jìn)行了封裝優(yōu)化，接下來(lái)，我們會(huì)繼續(xù)走近Netty，敬請(qǐng)期待。

參考：

[1].《Netty權(quán)威指南》

[2].https://juejin.cn/post/7130952602350534693#heading-14

[3].https://www.jianshu.com/p/670033e5b916

關(guān)鍵詞：