IO模型3-NIO(非阻塞IO)

1.NIO三大核心

1.1 Selecor选择器

• 事件驱动, 通过事件驱动那个通道有数据可读或者可写

• 一个selector对应一个线程, 可拥有多个Channel

1.2 Channel 通道

• 看做一个连接, 区别BIO基于流(读和写单向)， NIO基于通道(读写双向)

• 一个Channel对应一个连接, 拥有一个Buffer

1.3 Buffer 缓存区

• 是一个可读写的内存块, 说明NIO是读写是双向的

• 一个Buffer对应一个数据流(可读可写)

2. 非阻塞IO模型

2.1 NIO中的两个N

• 第一个N: java代码中的nio, 就是nio包下的代码

• 第二个N: nonblock, 内核机制-非阻塞机制

2.2 ServerSocketChannel和SocketChannel

1. ServerSocketChannel: 服务监听的socket, 监听有没有socket连接

   • 被动的, 被动的等待别人连接,

   • 然后监听到连接, 通过accept获取连接socket

   • 可将ServerSocketChannel看做BIO中使用的ServerSocket

2. SocketChannel: 服务连接的socket,

   • 主动的, 主动向客户端进行连接
   • 可将SocketChannel看做BIO中的Socket

3. 在linux的tcp连接上会出现两个socket, 一个的监听socket, 一个是连接socket

   • listen: 就是监听的socket, (ServerSocket)
   • establisthed: 就是连接成功的socket, (socket)

3. NIO的实现

3.1 未注册版

• 只设置非阻塞, 不做socket连接注册, 将连接存入一个list中, 通过循环list获取准备就绪的socket

• 重要点:

  1. 设置ServerSocketChannel的configuerBlocking为false, 意为’非阻塞’(默认阻塞)
  2. 设置SocketChannel的configuerBlocking为false, 意为’非阻塞’(默认阻塞)

		// 定义list存储socket通道
        LinkedList<SocketChannel> list = new LinkedList<>();
        // 打开服务监听socket通道
        ServerSocketChannel open = ServerSocketChannel.open();
        // 绑定端口, backlog限制连接数量为2
        open.bind(new InetSocketAddress("127.0.0.1",9091), 2);
        // 设置通道是否阻塞, true阻塞(BIO), false不阻塞(NIO)
        open.configureBlocking(false);
        // 死循环获取socket连接
        while (true) {
            // 监听socket通道,不阻塞,
            // (是一个连接到TCP网络套接字的通道,是一种面向流连接只sockets套接字的可选择通道)
            SocketChannel client = open.accept();
            // 判断连接是否为空, 连接为空什么都不做
            if (null == client) {
                // System.out.println("null");
            } else {
                // 连接不为空, 说明socket已经连接上了
                // 设置socket是否阻塞
                client.configureBlocking(false);
                System.out.println("socket端口: " + client.socket().getPort() + "已被连接");
                // 添加连接到list中
                list.add(client);
            }
            // 设置缓冲对象
            ByteBuffer buffer = ByteBuffer.allocateDirect(4096);
            // 循环判断socket连接通道里面的数据
            for (SocketChannel c : list) {
                // 读一个数据
                int num = c.read(buffer);
                // 判断数据,为0说明socket没有数据可读, 为-1则说明socket数据发送完毕, 只有大于0才有数据
                if (num > 0) {
                    // 设置buffer指针指向头, 因为写入后指针指向尾, 
                    // (读写转换, 将读buffer转换成写buffer)
                    buffer.flip();
                    // 定义一个byte数组, 大小为buffer.limit, buffer的极限位置
                    byte[] bytes = new byte[buffer.limit()];
                    buffer.get(bytes);
                    String s = new String(bytes);
                    System.out.println(s);
                    buffer.clear();
                }
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

3.2 注册版-配合使用Selector

• 使用步骤
  1. 创建一个ServerSocketChannel, 监听连接的Socket
  2. 绑定IP端口, 设置非阻塞模式
  3. 创建一个Selector, 将创建的ServerSocketChannel注册到Selector上,并设置监听模式(OP_ACCEPT)
  4. 循环等待客户端连接
  5. 获取所有注册的socket(selector.selectedKeys()), 判断事件发生(新连接,读写)
     • 如果是新连接就将新连接的socket注册到selector上
     • 如果是读写socket, 就处理数据
  6. 最后在selectKeys中移除移除当前事件(避免重复处理)

		// 创建一个监听socket
        ServerSocketChannel open = ServerSocketChannel.open();
        // 绑定端口
        open.bind(new InetSocketAddress(9092), 2);
        // 设置不阻塞
        open.configureBlocking(false);
        /* java提供的操作内核多路复用器的接口,
           select poll模型不开辟空间, epoll模型开辟空间, 优先选择epoll
           open相当于在系统内核中开辟了一块空间, 调用epoll_create */
        Selector selector = Selector.open();
        // 先将监听socket注册到selector, 并设置一个server模式, 这里是监听accept
        open.register(selector, SelectionKey.OP_ACCEPT);
        while (true) {
            // 阻塞方法, 当有事件准备就绪就返回, 如果返回值>0, 则说明有一个或者多个事件准备就绪
            // 可使用select()的其他方法,不阻塞selectNow()或者阻塞一定时间select(long timeout)
            selector.select();
            // 获取channel就绪事假列表
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = keys.iterator();
            while (iterator.hasNext()) {
                // 获拿出一个就绪事件
                SelectionKey key = iterator.next();
                if (key.isAcceptable()) {
                    //
                    /** 监听channel处理, 处理新的连接,将读事件注册到selector
                     *  如果是epoll就是在内核中写入SocketChannel*/
                    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                    SocketChannel accept = channel.accept();
                    accept.configureBlocking(false);
                    accept.register(selector, SelectionKey.OP_READ);
                } else if (key.isReadable()) {
                    /** 读channel处理,获取SocketChannel,完成数据处理 */
                    SocketChannel channel = (SocketChannel) key.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    int read = channel.read(buffer);
                    if (read > 0) {
                    	// 反转, 将读buffer转换为写buffer, 底层改变指针位置及buffer大小
                        buffer.flip();
                        byte[] bytes = new byte[buffer.limit()];
                        buffer.get(bytes);
                        String s = new String(bytes);
                        System.out.println(s);
                    }
                    channel.close();
                }
                // 在set中移除这个事件, 如果不移除, 那么这个事件会被重复处理
                iterator.remove();
            }
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49