Netty（四）- NIO三大组件之Selector

一、Selector 基本介绍

（1）Java 的 NIO，是非阻塞的 IO ，当用一个线程，处理多个客户端连接时，就会使用到 Selector（选择器）；
（2）Selector 能够检测到多个注册的通道上是否有事件发生（注意：多个 Channel 以事件的方式可以注册到同一个Selector上），如果有事件发生，便获取事件然后针对每个事件进行相应的处理。这样就可以只用一个单线程去管理多个通道，也就是管理多个连接和请求；
（3）只有在 连接/通道 真正有读写事件发生时，才会进行读写，这就大大减少了系统开销，并且不必为每个连接都创建一个线程，不用去维护多个线程，避免了多线程之间的上下文切换导致的开销；

二、Selector 特点

（1）Netty 的 IO 线程 NioEventLoop 聚合了 Selector（选择器，也叫多路复用器），可以同时并发处理成百上千个客户端连接。
（2）当线程从某客户端 Socket 通道进行读写数据时，若没有数据可用时，该线程可以处理其他任务。
（3）线程通常将非阻塞 IO 的空闲时间用于在其他通道上执行 IO 操作，所以单独的线程可以管理多个输入和输出通道。
（4）由于读写操作都是非阻塞的，这就可以充分提升 IO 线程的运行效率，避免由于频繁 I/O 阻塞导致的线程挂起。
（5）一个 I/O 线程可以并发处理 N 个客户端连接和读写操作，这从根本上解决了传统同步阻塞 I/O 一连接一线程模型，架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。

三、Selector类相关方法

Selector 类是一个抽象类, 常用方法和说明如下：

public abstract class Selector implements Closeable { 
	public static Selector open(); //得到一个选择器对象
	public int select(long timeout); //监控所有注册的通道，当其中有 IO 操作可以进行时，将对应的 SelectionKey 加入到内部集合中并返回，参数用来设置超时时间
	public Set<SelectionKey> selectedKeys(); //从内部集合中得到所有的 SelectionKey，SelectionKey是和Channel事件关联的	
}
selector.select() //阻塞，直到有事件发生
selector.select(1000); //阻塞 1000 毫秒，在 1000 毫秒后返回
selector.wakeup(); //唤醒 selector 
selector.selectNow(); //不阻塞，立马返还
1
2
3
4
5
6
7
8
9

四、NIO 非阻塞 网络编程原理分析图

NIO 非阻塞 网络编程相关的(Selector、SelectionKey、ServerScoketChannel 和 SocketChannel) 关系梳理图

对上图的说明:
（1）当客户端连接时，会通过 ServerSocketChannel 得到 SocketChannel；
（2）Selector 进行监听 select 方法，返回有事件发生的通道的个数；
（3）将 socketChannel 注册到 Selector 上，register(Selector selector, int options)，一个 selector 上可以注册多个 SocketChannel；
（4）注册后返回一个 SelectionKey，会和该 Selector 关联(加入到SelectionKey集合)；
（5）进一步得到各个 SelectionKey (有事件发生的SelectionKey)；
（6）再通过 SelectionKey的channel()方法，反向获取 SocketChannel ；
（7）可以通过得到的SocketChannel的事件，完成业务处理；

五、NIO 非阻塞 网络编程快速入门

编写一个 NIO 入门案例，实现服务器端和客户端之间的数据简单通讯（非阻塞）

• 服务端

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

        // 创建ServerSocketChannel
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        // 创建一个Selecor对象
        Selector selector = Selector.open();
        // 绑定一个端口6666, 在服务端监听
        serverSocketChannel.socket().bind(new InetSocketAddress(6666));
        // 设置为非阻塞
        serverSocketChannel.configureBlocking(false);
        // 把 serverSocketChannel 注册到  selector 关心的事件（OP_ACCEPT连接事件）
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        System.out.println("注册后的selectionkey 数量=" + selector.keys().size()); // 1

        // 循环等待客户端连接
        while (true) {

            if (selector.select(1000) == 0) {
                System.out.println("服务器等待了1秒，没有事件发生");
                continue;
            }

            /**
             *
             * 1.如果返回的>0，表示有事件发生
             * 2. selector.selectedKeys()返回事件的selectedKeys集合
             * 3.通过 selectionKeys 反向获取SocketChannel
             */
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            System.out.println("selectionKeys 数量 = " + selectionKeys.size());

            // 遍历 Set, 使用迭代器遍历
            Iterator<SelectionKey> keyIterator = selectionKeys.iterator();

            while (keyIterator.hasNext()) {
                // 获取到SelectionKey
                SelectionKey key = keyIterator.next();
                // 根据SelectionKey对应的SocketChannel中发生的事件做相应的处理
                if (key.isAcceptable()) { //如果是 OP_ACCEPT, 表示有新的客户端连接
                    // 给该客户端生成一个 SocketChannel
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    System.out.println("客户端连接成功 生成了一个 socketChannel " + socketChannel.hashCode());
                    // 将 SocketChannel 设置为非阻塞
                    socketChannel.configureBlocking(false);
                    // 将socketChannel 注册到selector, 关注的事件为 OP_READ，同时给socketChannel关联一个Buffer
                    socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
                    System.out.println("客户端连接后 ，注册的selectionkey 数量=" + selector.keys().size()); //2,3,4..
                }
                if (key.isReadable()) { //如果是 OP_READ, 表示有读事件发生
                    // 通过SelectionKey反向获取到对应的SocketChannel
                    SocketChannel channel = (SocketChannel) key.channel();
                    // 获取到该channel关联的buffer
                    ByteBuffer buffer = (ByteBuffer) key.attachment();
                    channel.read(buffer);
                    System.out.println("form 客户端 " + new String(buffer.array()));
                }
                // 手动从集合中移除当前的selectionKeym，防止重复操作
                keyIterator.remove();
                System.out.println("移除一个SelectionKey");
            }
        }
    }
}
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

• 客户端

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

        // 创建一个网络通道
        SocketChannel socketChannel = SocketChannel.open();
        // 设置非阻塞
        socketChannel.configureBlocking(false);
        // 提供服务器端的ip 和 端口
        InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 6666);
        // 连接服务器
        if (!socketChannel.connect(inetSocketAddress)) {
            while (!socketChannel.finishConnect()) {
                System.out.println("因为连接需要时间，客户端不会阻塞，可以做其它工作..");
            }
        }

        // 如果连接成功，就发送数据
        String str = "hello, 你好";
        // Wraps a byte array into a buffer，不需要手动设置字节数组大小了
        ByteBuffer buffer = ByteBuffer.wrap(str.getBytes());
        // 发送数据，将 buffer 数据写入 channel
        socketChannel.write(buffer);
        // 不让程序结束
        System.in.read();

    }
}
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

• 输出结果

注册后的selectionkey 数量=1
服务器等待了1秒，没有事件发生
服务器等待了1秒，没有事件发生
selectionKeys 数量 = 1
客户端连接成功 生成了一个 socketChannel 1254526270
客户端连接后 ，注册的selectionkey 数量=2
移除一个SelectionKey
selectionKeys 数量 = 1
form 客户端 hello, 你好
服务器等待了1秒，没有事件发生
服务器等待了1秒，没有事件发生
1
2
3
4
5
6
7
8
9
10
11

六、SelectionKey

SelectionKey，表示 Selector 和 网络通道 的注册关系, 共四种:

public static final int OP_READ = 1 << 0; // 代表读操作，值为 1
public static final int OP_WRITE = 1 << 2; // 代表写操作，值为 4
public static final int OP_CONNECT = 1 << 3; // 代表连接已经建立，值为 8
public static final int OP_ACCEPT = 1 << 4;// 有新的网络连接可以 accept，值为 16 
1
2
3
4

SelectionKey 相关方法

public abstract class SelectionKey {
 public abstract Selector selector(); //得到与之关联的 Selector 对象
 public abstract SelectableChannel channel(); //得到与之关联的通道
 public final Object attachment(); //得到与之关联的共享数据
 public abstract SelectionKey interestOps(int ops);//设置或改变监听事件
 public final boolean isAcceptable(); //是否可以accept
 public final boolean isReadable(); //是否可以读
 public final boolean isWritable(); //是否可以写
}
1
2
3
4
5
6
7
8
9