JAVA中的BIO,NIO与多路复用(select,poll,epoll)

1.Socket链接的建立

　　 java程序(server) 在应用空间中运行,当建立一个socket链接时,会向内核空间中的内核程序(sc)发送指令,内核程序中一定会执行Socket(AF_UNIX,SOCK_STAM,0) -> fd(文件标识符)6  --传递-> bind(6,9999) 绑定端口和文件标识符 --监听-->listen(6)

2.BIO (Blocking I/O)

　　概念:同步阻塞型IO,对于客户端的每个链接都将开启一个新的线程处理.

　　流程:在BIO中apccet的阻塞是在内核程序中阻塞,在使用strace 监控线程时可以看到 accept(6, 后停止打印,直到有链接建立然后才会继续剩余代码的执行. IO的阻塞在strace中可以看到是recv(5 停止打印直到有数据传输　　

　　优点:可以与多个客户端建立链接

　　缺点:一链接一线程,导致内存浪费.链接建立的越多,CPU线程切换更加频繁.

　　代码:

2 
import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.*;

/**
 * @author baiyang
 * @version 1.0
 * @date 2020/6/9 11:10 下午
 */
public class Server {

    public static void main(String[] args) throws Exception {

        ExecutorService executorService = Executors.newFixedThreadPool(50);
        final ServerSocket serverSocket = new ServerSocket(9999);
        while (true){
            Socket socket = serverSocket.accept(); // 阻塞
            new Thread(() ->{
                BufferedReader bufferedReader = null;
                int port = socket.getPort();
                System.out.println(port);
                try {
                    bufferedReader = new BufferedReader(new InputStreamReader(new BufferedInputStream(socket.getInputStream()))); // IO阻塞
                    String clientMessage = bufferedReader.readLine();
                    System.out.println(clientMessage);

                    BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(new BufferedOutputStream(socket.getOutputStream())));
                    bufferedWriter.write("this is service
");
                    bufferedWriter.flush();
                    bufferedWriter.close();
                    bufferedReader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();

        }
    }
}

package com.diandian.client.bio;

import java.io.*;
import java.net.InetSocketAddress;
import java.net.Socket;

/**
 * @author baiyang
 * @version 1.0
 * @date 2020/6/9 11:18 下午
 */
public class Client {

    public static void main(String[] args) throws Exception {
        Socket socket = new Socket();
        socket.connect(new InetSocketAddress(9090));
        BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(new BufferedOutputStream(socket.getOutputStream())));
        bufferedWriter.write("this is client
");
        bufferedWriter.flush();
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new BufferedInputStream(socket.getInputStream())));
        String clientMessage = bufferedReader.readLine();
        bufferedReader.close();
        bufferedWriter.close();
        System.out.println(clientMessage);


    }
}

3.NIO(java -> New I/O  操作系统 -> NONBLOCKING)

概念:在java中NIO 指的是new IO 1.4版本之后新的一个包出现

　　在操作系统中的体现是NONBLOCKING的支持

　　同步非阻塞

流程:在NIO中,ServerSocketChannel类中调用方法configureBlocking(false),accept将不在阻塞.当没有链接时accept() = -1,当客户端没有发送信息时recv() = -1.BIO中的阻塞就解决了

优点: 解决了BIO多线程的问题,解决了C10K问题

缺点: 循环遍历已链接的客户端,实现监控是否有数据写入. ---> 每一次的循环将会向内核程序发送一条指令,假设有1W个链接建立且只有1个客户端向服务器发送指令,那么需要向内核发送1W次指令,无效指令数9999次.

代码:

package com.diandian.server.nio;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;

/**
 * @author baiyang
 * @version 1.0
 * @date 2020/6/11 11:32 下午
 */
public class NioServer {

    public static void main(String[] args) {
        //  保存已链接的客户端
        ArrayList<SocketChannel> clients = new ArrayList<>();
        try {
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(9999));
            // 设置false不阻塞 调用 OS  NONBLOCKING
            serverSocketChannel.configureBlocking(false);
            while (true) {
                // 可有可无方便测试
                Thread.sleep(2000);
                SocketChannel client = serverSocketChannel.accept();
                if (null == client) {
                    System.out.println("没有客户端建立链接");
                } else {
                    client.configureBlocking(false);
                    clients.add(client);
                    System.out.println(client.socket().getPort());
                }
                ByteBuffer byteBuffer = ByteBuffer.allocateDirect(1024);
                for (SocketChannel s : clients){
                    int read = s.read(byteBuffer);
                    if(read > 0){
                        byteBuffer.flip();
                        byte[] bytes = new byte[byteBuffer.limit()];
                        byteBuffer.get(bytes);
                        System.out.println("收到客户端信息："+new String(bytes));
                        byteBuffer.clear();
                    }
                }

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

4.多路复用

 本质上select poll epoll 都是IO同步的,读写操作在就绪后都将由自己进行读写,所以读写操作是阻塞的

package com.diandian.service.nio;

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;
import java.util.Set;

/**
 * @author baiyang
 * @version 1.0
 * @date 2020/6/12 9:46 上午
 */
public class NioSelect {

    public static void main(String[] args) {
        try {
            ServerSocketChannel server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.bind(new InetSocketAddress(9999));
            // 开启selector
            Selector selector = Selector.open();
            // 将accept注册到selector
            server.register(selector,SelectionKey.OP_ACCEPT);
            while (true){
                System.out.println( " key Sizes:" + selector.selectedKeys().size());
                while (selector.select(1000) > 0){
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = keys.iterator();
                    while (iterator.hasNext()){
                        SelectionKey next = iterator.next();
                        // 移除防止重复遍历
                        iterator.remove();
                        // 是否有新的链接进来
                        if(next.isAcceptable()){
                            ServerSocketChannel channel = (ServerSocketChannel) next.channel();
                            SocketChannel cline = channel.accept();
                            // 设置非阻塞
                            cline.configureBlocking(false);
                            System.out.println(cline.socket().getPort());
                            ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
                            // 将read事件注册到selector中
                            cline.register(selector,SelectionKey.OP_READ,buffer);
                        }else if (next.isReadable()){
                            SocketChannel client = (SocketChannel) next.channel();
                            ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
                            buffer.clear();
                            int read;
                            while (true) {
                                read = client.read(buffer);
                                if (read > 0) {
                                    buffer.flip();
                                    while (buffer.hasRemaining()) {
                                        client.write(buffer);
                                    }
                                    buffer.clear();
                                } else if (read == 0) {
                                    break;
                                } else {
                                    client.close();
                                    break;
                                }
                            }
                        }

                    }
                }
            }

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

 

　　4.1 select poll

　　概念:NIO中解决多次向内核程序发送无效指令问题

　　流程: select(fd) --全量遍历--> 用户空间

　　优点:减少了向内核程序发送指令次数,一次将所有链接的fd发送给内核,由内核遍历

　　缺点:每次都将全量的fd进行发送,内核将进行全量遍历,只有一个selector来来进行监控accept() IO操作

　　select和poll的区别: 本质上是一致的, poll用链表存储无链接限制. select 监听fd有链接数量限制,LINUX32位默认 1024 64位默认 2048 

　　4.2 epoll (even poll)

概念:会将有发生IO操作的fd通知到应用程序
优点:不再每次进行全量的遍历 复杂度降低至O(1)可使用多个selector进行监控
流程: 建立链接后 epoll_create(256) -> 7 ----> epoll_crl(7,ADD,6(scoket建立时返回的文件标识符),accept) 将6的accept注册到7空间(红黑树保存fd)中 ----> epoll_wait() 阻塞的等待客户端的链接 O(1) -accept(6)->8 与客户端建立链接8 -> epoll_crl(7,ADD,8,read)将读事件注册到7空间 ----> epoll_wait(6,8)