3、Selector（选择器）

简单介绍：

Selector能够检测多个注册的通道上是否有事件发生（多个Channel以事件的方式可以注册到同一个Selector），如果有事件发生，便获取事件然后针对每个事件进行相应的处理，这样就可以只用一个单线程去管理多个通道。

只有在连接通道真正有读写事件发生时，才会进行读写，大大减少系统开销，不必为每个连接创建线程，同时避免了多线程之间上下文切换导致的开销。

特点：

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

Selector类和相关方法：

Selector是个抽象类，常用的方法：

public abstract class Selector implements Closeable {
    public static Selector open(); // 得到一个Selector对象
    public int select(long timeout); // 监控所有注册的通道，当其中有相关IO操作时，将对应的SelectionKey加入到内部集合中，timeout超时时间
    public int select(); // 阻塞的select操作
    public int selectNow(); // 非阻塞的select操作
    public abstract Selector wakeup(); // 唤醒正在阻塞的Selector 
    public Set<SelectionKey> selectedKeys(); // 返回当前的SelectionKey集合
}

4、NIO网络编程原理

NIO网络编程相关类（Selector，SelectionKey，ServerSocketChannel，SocketChannel）关系梳理：

1. ServerSocketChannel监听端口，当有客户端连接时生成对应的SocketChannel；ServerSocketChannel也要注册进Selector中，才能监听到客户端连接事件。
2. Selector负责监听，调用select() 方法，返回有事件发生的通道的个数。
3. 将SocketChannel注册到Selector上，注册方法register(Selector sel, int ops, Object att)，并返回SelecionKey，一个Selector上可注册多个SocketChannel。
4. 有注册的事件发生时，调用方法selectedKeys() 可获取当前SelecionKey集合。
5. 通过SelecionKey的channel() 方法可反向获取 SocketChannel，然后完成业务的处理
6. 事件类型：int OP_READ = 1 << 0; int OP_WRITE = 1 << 2; int OP_CONNECT = 1 << 3; int OP_ACCEPT = 1 << 4;

代码示例：

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.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;

/
 * NIO编程示例
 *
 * @author lzlg
 * 2020/8/9 11:41
 */
public class NIOServerClientDemo {
    /
     * NIO服务端
     */
    private static class NIOServer {
        public static void main(String[] args) {
            // 创建一个服务端的 ServerSocketChannel
            try (ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()) {
                // 绑定本地端口8888
                serverSocketChannel.socket().bind(new InetSocketAddress(8888));
                // 设置通道为非阻塞
                serverSocketChannel.configureBlocking(false);
                // 创建一个选择器
                Selector selector = Selector.open();
                // 将 ServerSocketChannel 注册进Selector中, 注册事件：接收连接 OP_ACCEPT
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

                while (true) {
                    // 每隔 1 秒检测是否有连接的事件发生
                    if (selector.select(1000) == 0) {
                        System.out.println("没有客户端连接, 等待 1 秒.....");
                        continue;
                    }
                    // 如果有连接事件发生, 则获取相关的 SelectionKey集合
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    // 使用迭代器遍历 SelectionKey集合
                    Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
                    while (keyIterator.hasNext()) {
                        SelectionKey selectionKey = keyIterator.next();
                        // 如果是可接收状态
                        if (selectionKey.isAcceptable()) {
                            // 则ServerSocketChannel接收客户端请求
                            SocketChannel socketChannel = serverSocketChannel.accept();
                            // 注意要设置为非阻塞
                            socketChannel.configureBlocking(false);
                            // 并将客户端请求获取到的 SocketChannel 注册入 Selector中
                            socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(512));
                        }
                        // 如果当前有SelectionKey是可读的
                        if (selectionKey.isReadable()) {
                            // 从 selectionKey 中获取 SocketChannel
                            SocketChannel channel = (SocketChannel) selectionKey.channel();
                            // 从 selectionKey 中获取 ByteBuffer
                            ByteBuffer buffer = (ByteBuffer) selectionKey.attachment();
                            // 将通道中的数据读取到Buffer中
                            channel.read(buffer);
                            System.out.println("从客户端获取到消息: " + new String(buffer.array()));
                        }

                        // 将迭代后的元素, 进行删除,
                        // 否则一个客户端连接后, serverSocketChannel的事件改变, 造成空指针
                        keyIterator.remove();
                    }

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

    /
     * NIO客户端
     */
    private static class NIOClient {
        public static void main(String[] args) {
            // 创建一个Socket通道
            try (SocketChannel socketChannel = SocketChannel.open()) {
                // 设置通道为非阻塞
                socketChannel.configureBlocking(false);
                // 设置服务器地址
                InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 8888);
                // 如果没有连接上, 则进行其他操作
                if (!socketChannel.connect(inetSocketAddress)) {
                    while (!socketChannel.finishConnect()) {
                        System.out.println("连接需要时间, 客户端不会阻塞, 可以做其他工作.....");
                    }
                }
                // 创建ByteBuffer
                String str = "Hello,李哲龙,张玉蒙";
                ByteBuffer buffer = ByteBuffer.wrap(str.getBytes(StandardCharsets.UTF_8));
                // 将ByteBuffer的数据读取到Channel中
                socketChannel.write(buffer);

                // 阻塞客户端
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

public abstract class SelectionKey {
    public abstract Selector selector(); // 得到与之关联的 Selector
    public abstract SelectableChannel channel(); // 得到与之关联的 Channel
    public final Object attachment(); // 得到与之关联的共享数据
    public abstract SelectionKey interestOps(int ops); // 设置或改变监听事件
    public final boolean isAcceptable(); // 是否可以 accept
    public final boolean isReadable(); // 是否可读
    public final boolean isWritable(); // 是否可写
}

public abstract class SeverSocketChannel extends AbstractSelectableChannel
    implements NetworkChannel {
    public static SeverSocketChannel open(); // 得到一个SeverSocketChannel
    public final SeverSocketChannel bind(SocketAddress local); // 设置服务器端口号
    public final SelectableChannel configureBlocking(boolean block); // 设置阻塞或非阻塞模式，取值false表示采用非阻塞模式
    public SocketChannel accept(); // 接收一个连接，返回代表这个连接的 SocketChannel
    public final SelectionKey register(Selector sel, int ops); // 注册一个选择器，并设置监听事件
}

public abstract class public abstract class SocketChannel
    extends AbstractSelectableChannel
    implements ByteChannel, ScatteringByteChannel, GatheringByteChannel, NetworkChannel {
    public static SocketChannel open(); // 得到一个SocketChannel
    public final SelectableChannel configureBlocking(boolean block); // 设置阻塞或非阻塞模式，取值false表示采用非阻塞模式
    public boolean connect(SocketAddress remote); // 连接服务器
    public boolean finishConnect(); // 如果上面的方法连接失败，接下来就要通过该方法完成连接操作
    public int write(ByteBuffer src); // 往通道中写数据
    public int read(ByteBuffer src); // 从通道中读数据
    public final SelectionKey register(Selector sel, int ops, Object att); // 注册一个选择器，并设置监听事件，最后一个参数可设置共享数据 
    public final void close(); // 关闭通道
}

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

/
 * 群聊系统服务端
 *
 * @author lzlg
 * 2020/8/15 18:47
 */
public class GroupChatServer {

    private Selector selector; // 选择器

    private ServerSocketChannel channel; // 服务器Socket通道

    private static final int PORT = 8675; // 服务器端口号

    /
     * 提供外部可访问的静态构造方法
     */
    public static GroupChatServer build() {
        return new GroupChatServer();
    }

    /
     * 构造器
     */
    private GroupChatServer() {
        try {
            // 创建选择器
            selector = Selector.open();
            // 创建 服务器Socket通道
            channel = ServerSocketChannel.open();
            // 绑定端口
            channel.socket().bind(new InetSocketAddress(PORT));
            // 设置非阻塞
            channel.configureBlocking(false);

            // 注册为Accept事件
            channel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("Server is start.....");

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

    /
     * 服务器启动
     * 1.监听客户端发送的消息
     * 2.将客户端发送的消息转发到其他客户端
     */
    public void run() {
        try {
            while (true) {
                // 监听选择器中发生的事件
                int select = selector.select();
                if (select > 0) {
                    // 获取所有的发生事件的SelectionKey
                    Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                    while (it.hasNext()) {
                        // 迭代 SelectionKey
                        SelectionKey key = it.next();
                        // 如果是接收事件，则打开客户端
                        if (key.isAcceptable()) {
                            // 获取监听客户端的Channel
                            SocketChannel channel = this.channel.accept();
                            // 设置为非阻塞，并注册为read事件
                            channel.configureBlocking(false);
                            channel.register(selector, SelectionKey.OP_READ);
                            // 标记客户端上线
                            System.out.println(channel.getRemoteAddress() + " 上线了.....");
                        }
                        // 如果是读取事件，则读取消息
                        if (key.isReadable()) {
                            // 获取到消息内容
                            String msg = read(key);

                            // 把消息转发到其他通道上
                            forward(msg, key);
                        }

                        // 注意这里要清除当前循环的key
                        it.remove();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();

        }
    }

    /
     * 读取消息
     *
     * @param key 有事件发生的通道
     */
    private String read(SelectionKey key) {
        SocketChannel channel = null;
        try {
            // 获取对应的通道
            channel = (SocketChannel) key.channel();
            // 使用8字节的buffer进行读取
            int read;
            ByteBuffer buffer = ByteBuffer.allocate(8);
            StringBuilder sb = new StringBuilder();
            while ((read = channel.read(buffer)) != 0) {
                sb.append(new String(buffer.array(), 0, read));
                // 这里必须进行rewind(),否则Buffer不能再次使用
                buffer.rewind();
            }
            // 获取到消息内容
            String msg = sb.toString();
            System.out.println("收到消息: " + msg);
            return msg;
        } catch (IOException e) {
            try {
                // client离线
                System.out.println(channel.getRemoteAddress() + " 离线了.....");
                // 取消注册
                key.cancel();
                // 关闭通道
                channel.close();
            } catch (IOException ex) {
                ex.printStackTrace();
            }
            return null;
        }
    }

    /
     * 将消息转发到其他通道上
     *
     * @param msg        消息内容
     * @param excludeKey 排除的通道
     */
    private void forward(String msg, SelectionKey excludeKey) throws IOException {
        if (msg == null) return;
        // 获取所有的通道
        Set<SelectionKey> keys = selector.keys();
        // 遍历
        for (SelectionKey key : keys) {
            SelectableChannel channel = key.channel();
            // 如果是SocketChannel, 且不为排除的通道
            if (channel instanceof SocketChannel && channel != excludeKey.channel()) {
                // 进行数据的写入
                SocketChannel socketChannel = (SocketChannel) channel;
                // 把Buffer中的数据写入通道中
                socketChannel.write(ByteBuffer.wrap(msg.getBytes()));
            }
        }
    }

    public static void main(String[] args) {
        GroupChatServer.build().run();
    }
}

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;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/
 * 群聊系统客户端
 *
 * @author lzlg
 * 2020/8/15 18:47
 */
public class GroupChatClient {

    private Selector selector; // 选择器

    private SocketChannel channel; // 通道

    private String name;

    private static final String SERVER_IP = "127.0.0.1";

    private static final int SERVER_PORT = 8675;

    private static final ExecutorService pool = Executors.newFixedThreadPool(4);

    /
     * 暴露给外部的构建方法
     */
    public static GroupChatClient build() {
        return new GroupChatClient();
    }

    private GroupChatClient() {
        try {
            // 创建选择器
            selector = Selector.open();
            // 创建Socket通道 通道绑定服务端ip和端口
            channel = SocketChannel.open(new InetSocketAddress(SERVER_IP, SERVER_PORT));
            // 设置非阻塞
            channel.configureBlocking(false);
            // 注册read事件
            channel.register(selector, SelectionKey.OP_READ);

            this.name = channel.getLocalAddress().toString().substring(1);
            System.out.println(this.name + " is ok.....");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /
     * 启动客户端
     */
    public void start() {
        // 使用线程池，执行读取消息的任务
        pool.execute(() -> {
            try {
                while (true) {
                    // 每一秒进行监听
                    int select = selector.select();
                    if (select > 0) {
                        Iterator<SelectionKey> it = selector.selectedKeys().iterator();

                        while (it.hasNext()) {
                            SelectionKey key = it.next();
                            if (key.isReadable()) {
                                SocketChannel channel = (SocketChannel) key.channel();
                                int read;
                                StringBuilder sb = new StringBuilder();
                                ByteBuffer buffer = ByteBuffer.allocate(8);
                                while ((read = channel.read(buffer)) != 0) {
                                    sb.append(new String(buffer.array(), 0, read));
                                    buffer.rewind();
                                }
                                System.out.println("读取到消息: " + sb.toString());
                            }
                        }

                        it.remove();
                    }

                    TimeUnit.SECONDS.sleep(1);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        try {
            // 主线程发送消息
            Scanner scanner = new Scanner(System.in);
            System.out.println("请输入: ");
            while (scanner.hasNextLine()) {
                String msg = this.name + " 说: " + scanner.nextLine();
                System.out.println(msg);
                this.channel.write(ByteBuffer.wrap(msg.getBytes()));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        GroupChatClient.build().start();
    }
}

/
 * 不知道怎么分类的小工具
 *
 * @author lzlg
 * 2020/8/16 11:33
 */
public class Tool {

    private Tool() {
    }

    /
     * 计算方法运行事件
     *
     * @param runnable 方法
     */
    public static void time(Runnable runnable) {
        long start = System.currentTimeMillis();
        runnable.run();
        long end = System.currentTimeMillis();
        System.out.println("运行时间(毫秒): " + (end - start));
    }
}

import java.io.DataInputStream;
import java.net.ServerSocket;
import java.net.Socket;

/
 * 传统IO服务端
 *
 * @author lzlg
 * 2020/8/16 11:40
 */
public class OldIOServer {
    public static void main(String[] args) throws Exception {
        ServerSocket serverSocket = new ServerSocket(8761);
        while (true) {
            Socket socket = serverSocket.accept();
            DataInputStream dis = new DataInputStream(socket.getInputStream());
            int read;
            byte[] bytes = new byte[1024];
            while ((read = dis.read(bytes)) != -1) {
//                System.out.println("old server is read. " + read);
            }
            dis.close();
        }
    }
}

import com.lzlg.study.util.Tool;

import java.io.DataOutputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;

/
 * 传统IO客户端
 *
 * @author lzlg
 * 2020/8/16 11:40
 */
public class OldIOClient {
    public static void main(String[] args) {
        Tool.time(() -> {
            try {
                Socket socket = new Socket();
                socket.connect(new InetSocketAddress("127.0.0.1", 8761));
                FileInputStream fis = new FileInputStream("activity-1.0.0.jar");
                byte[] bytes = new byte[1024];
                DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
                int read;
                while ((read = fis.read(bytes)) != -1) {
                    dos.write(bytes, 0, read);
//                    System.out.println("old client is write. " + read);
                }
                fis.close();
                dos.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }
}

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

/
 * 非阻塞IO服务端
 *
 * @author lzlg
 * 2020/8/16 11:40
 */
public class NewIOServer {
    public static void main(String[] args) throws Exception {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(9876));
        while (true) {
            SocketChannel socketChannel = serverSocketChannel.accept();
            int read;
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            while ((read = socketChannel.read(buffer)) != -1) {
                buffer.rewind();
            }
        }
    }
}

import com.lzlg.study.util.Tool;

import java.io.FileInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.FileChannel;
import java.nio.channels.SocketChannel;

/
 * 非阻塞IO客户端
 *
 * @author lzlg
 * 2020/8/16 11:40
 */
public class NewIOClient {
    public static void main(String[] args) {
        Tool.time(() -> {
            try {
                SocketChannel socketChannel = SocketChannel.open();
                socketChannel.connect(new InetSocketAddress("127.0.0.1", 9876));
                FileChannel fileChannel = new FileInputStream("activity-1.0.0.jar").getChannel();
                // windows环境下零拷贝只有8M
                final long M_8 = 8 * 1024 * 1024L;
                // 文件大小
                long size = fileChannel.size();
                // 计算传输次数
                long count = size / M_8 + 1;
                for (int i = 0; i < count; i++) {
                    fileChannel.transferTo(i * M_8, M_8, socketChannel);
                }
//                fileChannel.transferTo(0, fileChannel.size(), socketChannel);
                socketChannel.close();
                fileChannel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

    }
}