Java NIO的理解和應用_ZenDei技術網路在線

Java NIO是一種基於通道和緩衝區的I/O方式，已經被廣泛的應用，成為解決高併發與大量連接和I/O處理問題的有效方式。 Java NIO相關組件 Java NIO主要有三個核心部分組成，分別是：Channel(通道)，Buffer(緩衝區), Selector（選擇器） Channel Chan ...

Java NIO是一種基於通道和緩衝區的I/O方式，已經被廣泛的應用，成為解決高併發與大量連接和I/O處理問題的有效方式。

Java NIO相關組件

Java NIO主要有三個核心部分組成，分別是：Channel(通道)，Buffer(緩衝區), Selector（選擇器）

Channel

Channel是所有訪問IO設備的統稱。類型與IO中的Stream，而通道是雙向的，既可以讀又可以寫，但是Stream是單項的。常用的通道有：SocketChannel和ServerSocketChannel（對應TCP的客戶端和伺服器端）、FileChannel（對應文件IO）、DatagramChannel（對應UDP）等

Buffer

所有數據的讀寫都要經過Buffer，Buffer直接和Channel打交道，是一個存儲數據的容器。通過調用Channel.write方法將數據寫入Buffer，Channel.read方法將數據從Buffer中讀取出來。常用的Buffer有：ByteBuffer、LongBuffer、IntBuffer、StringCharBuffer等

Selector

Selector用來監聽多個Channel的事件（比如：Read、Write、Connect和Accept等），通過單個線程輪詢的方式實現了對多個Channel的監聽。

Java IO與NIO的區別

NIO是一種叫非阻塞IO（Non-blocking I/O），基於I/O多路復用來實現的（可參考：I/O模型、select、poll和epoll之間的區別）。NIO與之前傳統的I/O模型有很大的不同，具體表現在以下幾個方面：

面向流與面向緩衝

Java IO和NIO之間一個最大的區別是，IO是面向流的，NIO是面向緩衝區的。Java IO每次從數據流中讀一個或多個位元組，直至讀取所有位元組，數據流是一次性的，讀取完以後，不能前後移動流中的數據。Java NIO是將數據讀取到緩衝區，可以通過position來回移動訪問緩衝區中的數據。

阻塞與非阻塞IO

Java IO中調用read和write方法的線程會被阻塞的，直到數據全部讀入或者全部寫入完為止。而在Java NIO中，如果需要讀寫數據只用和緩衝區打交道，將數據從緩衝區讀取或者寫入緩衝區以後，線程可以繼續做其他事情，不會被block住。

選擇器（Selector）

Selector是基於I/O多路復用的機制實現的，將多個Channel註冊到一個Selector上，Selector通過輪詢監聽所有註冊的通道上是否有SelectionKey發生，如果發生了，然後將SelectionKey分派給其他線程處理。

Java NIO的應用

通過Java NIO技術簡單實現了一個服務端與客戶端通信的case，具體功能如下：

服務端可以向客戶端廣播消息
服務端將一個客戶端的消息轉發給其他客戶端
客戶端向服務端發送消息
客戶端接收服務端的消息

服務端代碼如下：

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Server {

    public static void main(String[] args) throws IOException {
        new Server().start(); // 啟動服務端程式
    }

    public Server() throws IOException {
        this.init(); // 初始化服務端數據
    }

    /**
     * 服務端埠
     */
    private int port = 9999;

    /**
     * 服務端的Selector用來監聽Channel的事件.
     */
    private Selector selector;

    /**
     * 字元數據編碼
     */
    private Charset charset = Charset.forName("UTF-8");

    /**
     * 讀緩存，分配1024Byte的空間
     */
    private ByteBuffer readBuffer = ByteBuffer.allocate(1024);

    /**
     * 寫緩存，分配1024Byte的空間
     */
    private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);

    /**
     * 存儲所有客戶端的Channel，轉發的時候使用
     */
    private Map<String, Channel> clientSocketChannels = new HashMap<>();

    /**
     * 定義了一個線程池，服務端用來發送數據給客戶端
     */
    private static ExecutorService executorService = Executors.newFixedThreadPool(1, runnable -> {
        Thread thread = new Thread(runnable);
        thread.setDaemon(true);
        thread.setName("server-sender");
        return thread;
    });

    /**
     * 初始化Channel.
     */
    private void init() throws IOException {
        // 聲明一個服務端的ServerSocketChannel
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        // 將服務端的ServerSocketChannel設置成非阻塞模式
        serverSocketChannel.configureBlocking(false);
        // 設置服務端的socket
        ServerSocket serverSocket = serverSocketChannel.socket();
        serverSocket.bind(new InetSocketAddress(this.port));
        // 聲明一個Selector，用來監聽服務端的所有Channel
        this.selector = Selector.open();
        // 在ServerSocketChannel上註冊Accept事件，用來接收客戶端的連接
        serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
        System.out.println("Server is started, the port is " + this.port);
    }

    /**
     * 處理服務端監聽到的事件
     */
    private void work(SelectionKey selectionKey) throws IOException {
        // 客戶端有Socket連接請求
        if (selectionKey.isAcceptable()) {
            // 從SelectionKey中獲取服務端的ServerSocketChannel，SelectionKey中包含了服務端與客戶端的所有信息
            ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
            // 服務端打開一個新的SocketChannel用來與客戶端的SocketChannel進行通信，服務端同時會隨機分配一個埠
            SocketChannel socketChannel = serverSocketChannel.accept();
            // 將SocketChannel設置成非阻塞模式
            socketChannel.configureBlocking(false);
            // 將SocketChannel中的Read事件註冊到Selector上
            socketChannel.register(this.selector, SelectionKey.OP_READ);
            // 存儲服務端為客戶端創建的SocketChannel，為後面的轉發消息服務
            this.clientSocketChannels.put(this.getClientName(socketChannel), socketChannel);
            // 通過System.in IO流來創建Scanner
            Scanner scanner = new Scanner(System.in);
            // 收集服務端控制台輸入的數據，通過線程池將數據廣播給所有客戶端SocketChannel
            this.executorService.submit(() -> {
                while (true) {
                    // 該方法會被block住，一直等到服務端控制台有數據輸入完為止
                    String sendText = scanner.nextLine();
                    // 將服務端的數據廣播給所有客戶端
                    transferToOthers(sendText, null);
                }
            });
        // 服務端監聽到有數據可以讀取，主要是來源於客戶端發送的數據
        } else if (selectionKey.isReadable()) {
            // 獲取服務端的SocketChannel，然後與客戶端進行通信
            // 需要註意的是：當前獲取的SocketChannel與ServerSocketChannel是不同的，
            // 這個SocketChannel是通過調用ServerSocketChannel.accept方法創建的（存儲在clientSocketChannels集合中）
            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
            // 清空當前的用來存儲讀數據的buffer
            readBuffer.clear();
            // 將數據從SocketChannel讀入buffer
            int bytes = socketChannel.read(readBuffer);
            if (bytes > 0) {
                // 使得buffer中的數據可讀
                readBuffer.flip();
                // 讀取buffer中的數據
                String text = String.valueOf(this.charset.decode(readBuffer));
                System.out.println(this.getClientName(socketChannel) + ": " + text);
                // 將客戶端發送過來的數據轉發給其他客戶端
                this.transferToOthers(text, socketChannel);
            }
        }
    }

    /**
     * 將數據發送給其他客戶端
     */
    private void transferToOthers(String text, final SocketChannel socketChannel) {
        this.clientSocketChannels.forEach((channelName, channel) -> {
            // 獲取之前存儲的與服務端建立連接的客戶端
            SocketChannel otherSocketChannel = (SocketChannel) channel;
            if (!otherSocketChannel.equals(socketChannel)) {
                // 清空寫緩存
                this.writeBuffer.clear();
                // 將數據寫入緩存
                this.writeBuffer.put(this.charset.encode(this.getClientName(socketChannel) + ": " + text));
                // 使得緩存中的數據變得可用
                this.writeBuffer.flip();
                try {
                    // 將buffer中的數據寫入到其它客戶端
                    otherSocketChannel.write(this.writeBuffer);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 通過SocketChannel生成客戶端的名字，用來標識
     */
    private String getClientName(SocketChannel socketChannel) {
        if (socketChannel == null)
            return "[server]";
        Socket socket = socketChannel.socket();
        return "[" + socket.getInetAddress().toString().substring(1) + ":" + socket.getPort() + "]";
    }

    /**
     * 啟動服務端程式
     */
    public void start() {
        // 無限迴圈來輪詢所有註冊的Channel
        while (true) {
            try {
                // 選擇已經準備好的Channel，該方法是會block住的，直到有事件到達
                this.selector.select();
                // 獲取所有監聽到的事件
                Iterator<SelectionKey> iterator = this.selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    // 找到事件SelectionKey，裡面包含了事件相關的所有數據
                    SelectionKey selectionKey = iterator.next();
                    // 如果事件是有效的
                    if (selectionKey.isValid()) {
                        // 處理事件
                        this.work(selectionKey);
                    }
                    // 刪除已經處理過的事件
                    iterator.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

客戶端代碼如下：

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

public class Client {

    public static void main(String[] args) throws IOException {
        new Client().start(); // 客戶端程式執行入口
    }

    /**
     * 註冊監聽的服務的埠，並初始化
     */
    public Client() throws IOException {
        this.serverSocketAddress = new InetSocketAddress("127.0.0.1", 9999);
        this.init();
    }

    /**
     * 服務的Socket地址
     */
    private SocketAddress serverSocketAddress;

    /**
     * 客戶端Selector
     */
    private Selector selector;

    /**
     * 字元編碼
     */
    private Charset charset = Charset.forName("UTF-8");

    /**
     * 讀緩衝區
     */
    private ByteBuffer readBuffer = ByteBuffer.allocate(1024);

    /**
     * 寫緩衝區
     */
    private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);

    /**
     * 線程池執行客戶端發送數據
     */
    private static ExecutorService executorService = Executors.newFixedThreadPool(1, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(runnable);
            thread.setDaemon(true);
            thread.setName("client-sender");
            return thread;
        }
    });

    /**
     * 初始化客戶端信息
     */
    private void init() throws IOException {
        // 聲明一個客戶端SocketChannel
        SocketChannel socketChannel = SocketChannel.open();
        // 設置成非阻塞模式
        socketChannel.configureBlocking(false);
        // 聲明一個Selector
        this.selector = Selector.open();
        // 將客戶端的SocketChannel的連接事件註冊到selector上
        socketChannel.register(this.selector, SelectionKey.OP_CONNECT);
        // 連接服務端
        socketChannel.connect(this.serverSocketAddress);
    }

    /**
     * 處理客戶端數據
     */
    private void work(SelectionKey selectionKey) {
        try {
            //  與服務端建立連接
            if (selectionKey.isConnectable()) {
                // 從SelectionKey中獲取客戶端的ServerSocketChannel
                SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                // 判斷連接是否完成
                if (socketChannel.isConnectionPending()) {
                    // 完成連接
                    socketChannel.finishConnect();
                    System.out.println("The connection is successful！");
                    // 通過System.in IO流來創建Scanner
                    Scanner scanner = new Scanner(System.in);
                    // 使用線程池來完成對客戶端的控制台數據輸入的監聽
                    executorService.submit((Runnable) () -> {
                        while (true) {
                            try {
                                // 清空寫緩衝區
                                writeBuffer.clear();
                                // 該方法會被block住，一直等到客戶端控制台有數據輸入完為止
                                String sendText = scanner.nextLine();
                                // 將數據寫入寫緩衝區
                                writeBuffer.put(charset.encode(sendText));
                                // 使得寫緩衝區中的數據可讀
                                writeBuffer.flip();
                                // 將數據通過SocketChannel發送到服務端
                                socketChannel.write(writeBuffer);
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    });
                }
                // 註冊可讀事件，應該當前的SocketChannel與服務端建立連接以後，不需要再監聽創建連接的事件
                // 為了復用SocketChannel，將SocketChannel的Read事件註冊到Selector
                socketChannel.register(selector, SelectionKey.OP_READ);
            }
            // 可讀事件，有從伺服器端發送過來的信息，讀取輸出到控制臺上
            else if (selectionKey.isReadable()) {
                // 獲取與服務端通信的客戶端SocketChannel
                SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                // 清空讀緩衝區
                this.readBuffer.clear();
                // 將數據讀取到讀緩衝區，並將數據輸出到客戶端控制台
                int count = socketChannel.read(this.readBuffer);
                if (count > 0) {
                    String text = new String(this.readBuffer.array(), 0, count);
                    System.out.println(text);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 啟動客戶端程式
     */
    public void start() throws IOException {
        // 無限迴圈，輪詢所有監聽的SocketChannel
        while (true) {
            // 選擇已經準備好的Channel，該方法是會block住的，直到有事件到達
            int events = this.selector.select();
            if (events > 0) {
                // 找到事件SelectionKey，裡面包含了事件相關的所有數據
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                // 處理事件
                selectionKeys.forEach(selectionKey -> this.work(selectionKey));
                // 清空已處理的事件
                selectionKeys.clear();
            }
        }
    }
}

總結

服務端的ServerSocketChannel是用來監聽客戶端的連接請求，只有1個且埠固定，主要監聽accept事件
服務端的SocketChannel是用來和客戶端建立數據讀寫操作通信，數量與客戶端的連接數量一致，每個都分配一個隨機的埠，主要監聽read事件
每個客戶端有一個SocketChannel，用來和服務端進行通信，主要監聽connect事件和read事件，connect事件只會在第一連接時發生，read事件是在每次接收服務端數據時發生
服務端和客戶端各有一個Selector，用來監聽所有的SocketChannel或者ServerSocketChannel中註冊的事件，在沒有事件發生的時候，Selector.select()會被block住
在定義緩衝區的時候要註意緩衝區的大小，如果太小會報BufferOverflowException