Java NIO深入了解ServerSocketChannel

Java NIO 簡介

JAVA NIO有兩種解釋：一種叫非阻塞IO（Non-blocking I/O），另一種也叫新的IO（New I/O），其實是同一個概念。它是一種同步非阻塞的I/O模型，也是I/O多路複用的基礎，已經被越來越多地應用到大型應用伺服器，成為解決高并發與大量連接配接、I/O處理問題的有效方式。

NIO是一種基于通道和緩沖區的I/O方式，它可以使用Native函數庫直接配置設定堆外記憶體（差別于JVM的運作時資料區），然後通過一個存儲在java堆裡面的DirectByteBuffer對象作為這塊記憶體的直接引用進行操作。這樣能在一些場景顯著提高性能，因為避免了在Java堆和Native堆中來回複制資料。

Java NIO元件

NIO主要有三大核心部分：Channel(通道)，Buffer(緩沖區), Selector（選擇器）。傳統IO是基于位元組流和字元流進行操作（基于流），而NIO基于Channel和Buffer(緩沖區)進行操作，資料總是從通道讀取到緩沖區中，或者從緩沖區寫入到通道中。Selector(選擇區)用于監聽多個通道的事件（比如：連接配接打開，資料到達）。是以，單個線程可以監聽多個資料通道。

Buffer

Buffer（緩沖區）是一個用于存儲特定基本類型資料的容器。除了boolean外，其餘每種基本類型都有一個對應的buffer類。Buffer類的子類有ByteBuffer, CharBuffer, DoubleBuffer, FloatBuffer, IntBuffer, LongBuffer, ShortBuffer 。

Channel

Channel（通道）表示到實體，如硬體裝置、檔案、網絡套接字或可以執行一個或多個不同 I/O 操作（如讀取或寫入）的程式元件的開放的連接配接。Channel接口的常用實作類有FileChannel（對應檔案IO）、DatagramChannel（對應UDP）、SocketChannel和ServerSocketChannel（對應TCP的用戶端和伺服器端）。Channel和IO中的Stream(流)是差不多一個等級的。隻不過Stream是單向的，譬如：InputStream, OutputStream.而Channel是雙向的，既可以用來進行讀操作，又可以用來進行寫操作。

Selector

Selector（選擇器）用于監聽多個通道的事件（比如：連接配接打開，資料到達）。是以，單個的線程可以監聽多個資料通道。即用選擇器，借助單一線程，就可對數量龐大的活動I/O通道實施監控和維護。

Java NIO的簡單實作

public class Demo1 {

    private static Integer port = 8080;

    // 通道管理器(Selector)
    private static Selector selector;

    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 10, 1000, TimeUnit.MILLISECONDS, new LinkedTransferQueue<>(), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) {
        try {
            // 建立通道ServerSocketChannel
            ServerSocketChannel open = ServerSocketChannel.open();

            // 将通道設定為非阻塞
            open.configureBlocking(false);

            // 綁定到指定的端口上
            open.bind(new InetSocketAddress(port));

            // 通道管理器(Selector)
            selector = Selector.open();

            /**
             * 将通道(Channel)注冊到通道管理器(Selector)，并為該通道注冊selectionKey.OP_ACCEPT事件
             * 注冊該事件後，當事件到達的時候，selector.select()會傳回，
             * 如果事件沒有到達selector.select()會一直阻塞。
             */
            open.register(selector, SelectionKey.OP_ACCEPT);

            // 循環處理
            while (true) {
                /**
                 * 當注冊事件到達時，方法傳回，否則該方法會一直阻塞
                 * 該Selector的select()方法将會傳回大于0的整數，該整數值就表示該Selector上有多少個Channel具有可用的IO操作
                 */
                int select = selector.select();
                System.out.println("目前有 " + select + " 個channel可以操作");

                // 一個SelectionKey對應一個就緒的通道
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    // 擷取事件
                    SelectionKey key = iterator.next();

                    // 移除事件，避免重複處理
                    iterator.remove();

                    // 用戶端請求連接配接事件，接受用戶端連接配接就緒
                    if (key.isAcceptable()) {
                        accept(key);
                    } else if (key.isReadable()) {
                        // 監聽到讀事件，對讀事件進行處理
                        threadPoolExecutor.submit(new NioServerHandler(key));
                    }
                }

            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 處理用戶端連接配接成功事件
     *
     * @param key
     */
    public static void accept(SelectionKey key) {
        try {
            // 擷取用戶端連接配接通道
            ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
            SocketChannel sc = ssc.accept();
            sc.configureBlocking(false);

            // 給通道設定讀事件，用戶端監聽到讀事件後，進行讀取操作
            sc.register(selector, SelectionKey.OP_READ);
            System.out.println("accept a client : " + sc.socket().getInetAddress().getHostName());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 監聽到讀事件，讀取用戶端發送過來的消息
     */
    public static class NioServerHandler implements Runnable {

        private SelectionKey selectionKey;

        public NioServerHandler(SelectionKey selectionKey) {
            this.selectionKey = selectionKey;
        }

        @Override
        public void run() {
            try {
                if (selectionKey.isReadable()) {
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();

                    // 從通道讀取資料到緩沖區
                    ByteBuffer buffer = ByteBuffer.allocate(1024);

                    // 輸出用戶端發送過來的消息
                    socketChannel.read(buffer);
                    buffer.flip();
                    System.out.println("收到用戶端" + socketChannel.socket().getInetAddress().getHostName() + "的資料：" + new String(buffer.array()));

                    //将資料添加到key中
                    ByteBuffer outBuffer = ByteBuffer.wrap(buffer.array());

                    // 将消息回送給用戶端
                    socketChannel.write(outBuffer);
                    selectionKey.cancel();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
           

原文位址

https://www.51csdn.cn/article/404.html