【從入門到放棄-Java】併發編程-NIO使用

時間 2019-11-16

原文原文鏈接

前言

上文【從入門到放棄-SpringBoot】SpringBoot源碼分析-請求過程中咱們瞭解到，tomcat接收、返回請求的過程都是基於NIO實現的。平常工做中有不少基於NIO的使用，咱們知道NIO能夠提升系統的併發度，接下來的系列咱們來深刻學習下NIO，本文先從使用上簡單概述。html

NIO概述

NIO即non-blocking（New IO），是指jdk1.4 及以上版本里提供的新api。java

NIO和IO最大的區別：IO是以流的方式處理數據，而NIO是以塊的方式處理數據；IO對事件的處理是阻塞的，NIO是非阻塞的api

NIO的核心部分：tomcat

Channel
Buffer
Selector

NIO主要分爲標準輸入輸出和網絡請求網絡

標準輸入輸出NIO

讀取

private static void readNio() {
    try {
        //一、開啓文件讀取流
        FileInputStream fileInputStream = new FileInputStream("/Users/my/Desktop/123.txt");

        //二、獲取fileChannel
        FileChannel channel = fileInputStream.getChannel();

        //三、設置ByteBuffer大小，一次能容納capacity字節
        int capacity = 9;
        ByteBuffer bf = ByteBuffer.allocate(capacity);

        //四、當read返回-1時，表示文件讀取完畢
        int length = -1;
        while ((length = channel.read(bf)) != -1) {

            byte[] bytes = bf.array();
            System.out.println(new String(bytes, 0, length));

            //四、將bf position置爲0，方便下次讀取
            bf.clear();

        }
        channel.close();
        fileInputStream.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

寫入

private static void writeNio() {
    try {
        //一、打開文件寫入流
        FileOutputStream fileOutputStream = new FileOutputStream("/Users/my/Desktop/123.txt");

        //二、獲取fileChannel
        FileChannel channel = fileOutputStream.getChannel();

        //三、初始化byteBuffer
        String str = "薩達案發生大大sdada34;sdds'";
        ByteBuffer bf = ByteBuffer.allocate(1024);

        //四、將bf position置爲0，方便下次讀取
        bf.clear();

        //五、從byteBuffer的position位置填充byte
        bf.put(str.getBytes());

        //六、將bf position置爲0，limit設置爲position避免寫入內容過多
        bf.flip();

        int length = 0;

        //七、若是position小於limit即未寫入完畢
        while (bf.hasRemaining()) {
            //八、將buffer內容寫入channel
            length = channel.write(bf);
            System.out.println(bf);
        }
        channel.close();
        fileOutputStream.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

網絡NIO

服務端

package com.my.tools.nio;

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;

public class ServerSocket {
    private static ServerSocket serverSocket;

    private Selector selector;

    public static void main(String[] args) throws Exception {
        ServerSocket.getInstance().init(8001).listen();
    }

    public static ServerSocket getInstance() {
        if (serverSocket == null) {
            synchronized (ServerSocket.class) {
                if (serverSocket == null) {
                    serverSocket = new ServerSocket();
                }
            }
        }
        return serverSocket;
    }

    public ServerSocket init(int port) throws IOException {
        //初始化channel
        ServerSocketChannel server = ServerSocketChannel.open();

        //綁定本機8001端口
        server.socket().bind(new InetSocketAddress(8001));

        //設置爲非阻塞模式
        server.configureBlocking(false);

        //開啓selector管理器
        selector = Selector.open();

        //將selector註冊至server，並設置只處理accept事件
        server.register(selector, SelectionKey.OP_ACCEPT);

        return this;
    }

    public void listen() throws Exception {
        System.out.println("server start");

        //無限循環持續監聽
        while (true) {
            //會阻塞 直到監聽到註冊的事件
            selector.select();

            //獲取喚醒的事件
            Iterator<SelectionKey> selectorKeys = selector.selectedKeys().iterator();

            while (selectorKeys.hasNext()) {
                SelectionKey key = selectorKeys.next();

                //將已取出的SelectionKey刪除，防止重複處理
                selectorKeys.remove();

                if (key.isAcceptable()) {

                    //獲取到服務端的socket
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();

                    //獲取接收到的客戶端socket
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);

                    //向客戶端寫消息
                    socketChannel.write(ByteBuffer.wrap(new String("hello, this is server").getBytes()));

                    //註冊監聽read事件
                    socketChannel.register(selector, SelectionKey.OP_READ);
                    System.out.println("accept");
                } else if (key.isReadable()) {
                    //使用selector獲取channel
                    SocketChannel socketChannel = (SocketChannel) key.channel();
                    socketChannel.configureBlocking(false);

                    ByteBuffer buffer = ByteBuffer.allocate(1024);

                    //讀消息
                    int length = socketChannel.read(buffer);

                    String string = new String(buffer.array(), 0 , length);

                    System.out.println("read:" + socketChannel + string);

                    //寫消息
                    socketChannel.write(ByteBuffer.wrap(("server " + System.currentTimeMillis()).getBytes()));
                    Thread.sleep(10000);
                }
            }
        }
    }

}

客戶端

package com.my.tools.nio;

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;

public class ClientSocket {
    public static ClientSocket clientSocket;

    private static Selector selector;

    public static void main(String[] args) throws Exception {
        ClientSocket.getInstance().init("localhost", 8001).listen();
    }

    public static ClientSocket getInstance() {
        if (clientSocket == null) {
            synchronized (ClientSocket.class) {
                if (clientSocket == null) {
                    clientSocket = new ClientSocket();
                }
            }
        }

        return clientSocket;
    }

    public ClientSocket init(String ip, int port) throws IOException {
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(new InetSocketAddress(ip, port));
        socketChannel.configureBlocking(false);

        selector = Selector.open();
        socketChannel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);

        return this;
    }

    public void listen() throws Exception {
        System.out.println("client start");

        while (true) {
            selector.select();

            Iterator<SelectionKey> selectionKeys = selector.selectedKeys().iterator();

            while (selectionKeys.hasNext()) {
                SelectionKey selectionKey = selectionKeys.next();
                selectionKeys.remove();

                if (selectionKey.isConnectable()) {
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    socketChannel.configureBlocking(false);

                    ByteBuffer buffer = ByteBuffer.wrap(new String("hello, this is client").getBytes());
                    socketChannel.write(buffer);

                    socketChannel.register(selector, SelectionKey.OP_READ);
                    System.out.println("client write");
                } else if (selectionKey.isReadable()) {
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    socketChannel.configureBlocking(false);

                    ByteBuffer buffer = ByteBuffer.allocate(1024);

                    int length = socketChannel.read(buffer);

                    System.out.println("client read: " + socketChannel + new String(buffer.array(), 0, length));

                    socketChannel.write(ByteBuffer.wrap(("client " + System.currentTimeMillis()).getBytes()));

                    Thread.sleep(10000);
                }
            }

        }
    }
}