7、NIO：手撸一个简易的主从多Reactor线程模型

private static class Acceptor implements Runnable {
        // main Reactor 线程池，用于处理客户端的连接请求
        private static ExecutorService mainReactor = Executors.newSingleThreadExecutor(new ThreadFactory() {
            private AtomicInteger num = new AtomicInteger(0);
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                // 为线程池中的名称进行命名，方便分析线程栈
                t.setName("main-reactor-" + num.incrementAndGet());
                return t;
            }
        });
        public void run() {
            // NIO中服务端对应的Channel
            ServerSocketChannel ssc = null;
            try {
              // 通过静态方法创建一个ServerSocketChannel对象
                ssc = ServerSocketChannel.open();
              //设置为非阻塞模式
                ssc.configureBlocking(false);
              //绑定端口
                ssc.bind(new InetSocketAddress(SERVER_PORT));
​
                //转发到 MainReactor反应堆
                dispatch(ssc);
                System.out.println("服务端成功启动。。。。。。");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        private void dispatch(ServerSocketChannel ssc) {
            mainReactor.submit(new MainReactor(ssc));
        }
    }

public class MainReactor implements Runnable{
    // NIO 事件选择器
    private Selector selector;
    // 子ReactorThreadGroup 即IO线程池
    private SubReactorThreadGroup subReactorThreadGroup;
    // IO线程池默认线程数量
    private static final int DEFAULT_IO_THREAD_COUNT = 4;
    // IO线程个数
    private int ioThreadCount = DEFAULT_IO_THREAD_COUNT;
  
    public MainReactor(ServerSocketChannel channel) {
        try {
            // 创建事件选择器
            selector = Selector.open();
            // 为通道注册OP_ACCEPT 事件，客户端发送数据后，服务端通过该事件进行数据的读取
            channel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        // IO线程池，里面包含负载均衡算法
        subReactorThreadGroup = new SubReactorThreadGroup(ioThreadCount);
    }
    public void run() {
        System.out.println("MainReactor is running");
        while (!Thread.interrupted()) {
            Set<SelectionKey> ops = null;
            try {
                // 进行事件选择
                selector.select(1000);
                // 经过事件选择后已经就绪的事件
                ops = selector.selectedKeys(); 
            } catch (IOException e) {
                e.printStackTrace();
            }
            // 处理相关事件
            for (Iterator<SelectionKey> it = ops.iterator(); it.hasNext();) {
                SelectionKey key = it.next();
                it.remove();
                try {
                     // 如果有客户端尝试建立连接
                    if (key.isAcceptable()) {
                        System.out.println("收到客户端的连接请求。。。");
                        //获取服务端的ServerSocketChannel对象， 这里其实，可以直接使用ssl这个变量
                        ServerSocketChannel serverSc = (ServerSocketChannel) key.channel();
                        // 调用ServerSocketChannel的accept方法，创建SocketChannel
                        SocketChannel clientChannel = serverSc.accept();
                        // 设置为非阻塞模式
                        clientChannel.configureBlocking(false);
                        // 转发到IO线程，由对应的IO线程去负责网络读写
                        subReactorThreadGroup.dispatch(clientChannel); // 转发该请求
                    }
                } catch (Throwable e) {
                    e.printStackTrace();
                    System.out.println("客户端主动断开连接。。。。。。。");
                }
            }
        }
    }
}

public class SubReactorThreadGroup {
    private static final AtomicInteger requestCounter = new AtomicInteger();  //请求计数器
    // 线程池IO线程的数量
    private final int ioThreadCount;  
    // 业务线程池大小
    private final int businessTheadCout; 
    private static final int DEFAULT_NIO_THREAD_COUNT;
    // IO线程池数组
    private SubReactorThread[] ioThreads;
    //业务线程池
    private ExecutorService businessExecutePool; 
​
    static {
        DEFAULT_NIO_THREAD_COUNT = 4;
    }
​
    public SubReactorThreadGroup() {
        this(DEFAULT_NIO_THREAD_COUNT);
    }
    public SubReactorThreadGroup(int ioThreadCount) {
        if(ioThreadCount < 1) {
            ioThreadCount = DEFAULT_NIO_THREAD_COUNT;
        }
        //暂时固定为10
        businessTheadCout = 10;
        //初始化代码
        businessExecutePool = Executors.newFixedThreadPool(businessTheadCout, new ThreadFactory() {
            private AtomicInteger num = new AtomicInteger(0);
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("bussiness-thread-" + num.incrementAndGet());
                return t;
            }
        });
        this.ioThreadCount = ioThreadCount;
        this.ioThreads = new SubReactorThread[ioThreadCount];
        for(int i = 0; i < ioThreadCount; i ++ ) {
            this.ioThreads[i] = new SubReactorThread(businessExecutePool);
            this.ioThreads[i].start(); //构造方法中启动线程，由于nioThreads不会对外暴露，故不会引起线程逃逸
        }
        System.out.println("Nio 线程数量：" + ioThreadCount);
    }
    public void dispatch(SocketChannel socketChannel) {
         //根据负载算法转发到具体IO线程
        if(socketChannel != null ) {
            next().register(new NioTask(socketChannel, SelectionKey.OP_READ));
        }
    }
    protected SubReactorThread next() {
        return this.ioThreads[ requestCounter.getAndIncrement() %  ioThreadCount ];
    }

public class SubReactorThread extends Thread{
  // 事件选择器
    private Selector selector;
  //业务线程池
    private ExecutorService businessExecutorPool;
  //任务列表
    private List<NioTask> taskList = new ArrayList<NioTask>(512);
  // 锁
    private ReentrantLock taskMainLock = new ReentrantLock();
    /**
     * 业务线程池
     * @param businessExecutorPool
     */
    public SubReactorThread(ExecutorService businessExecutorPool) {
        try {
            this.businessExecutorPool = businessExecutorPool;
            //创建事件选择器
            this.selector = Selector.open();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    /**
     * 接受读写任务
     *
     */
    public void register(NioTask task) {
        if (task != null) {
            try {
                taskMainLock.lock();
                taskList.add(task);
            } finally {
                taskMainLock.unlock();
            }
        }
    }
    /**
     * 此处的reqBuffer处于可写状态
     * @param sc
     * @param reqBuffer
     */
    private void dispatch(SocketChannel sc, ByteBuffer reqBuffer) {
        businessExecutorPool.submit( new Handler(sc, reqBuffer, this)  );
    }
​
​
​
    public void run() {
        while (!Thread.interrupted()) {
            Set<SelectionKey> ops = null;
            try {
                //执行事件选择
                selector.select(1000);
              // 获取已就绪的事件集合
                ops = selector.selectedKeys();
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                continue;
            }
            // 处理相关事件
            for (Iterator<SelectionKey> it = ops.iterator(); it.hasNext();) {
                SelectionKey key = it.next();
                it.remove();
                try {
                     // 通道写事件就绪，说明可以继续往通道中写数据
                    if (key.isWritable()) { 
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        // 获取上次未写完的数据
                        ByteBuffer buf = (ByteBuffer) key.attachment();
                        // 将其写入到通道中。
                        // 这里实现比较粗糙，需要采用处理taskList类似的方式，因为此时通道缓冲区有可能已写满
                        clientChannel.write(buf);
                        System.out.println("服务端向客户端发送数据。。。");
                        // 重新注册读事件
                        clientChannel.register(selector, SelectionKey.OP_READ);
                    } else if (key.isReadable()) { // 接受客户端请求
                        System.out.println("服务端接收客户端连接请求。。。");
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        ByteBuffer buf = ByteBuffer.allocate(1024);
                        System.out.println(buf.capacity());
                        /**
                         * 这里其实实现的非常不优雅，需要对读取处理办关闭，而且一次读取，并不一定能将一个请求读取
                         * 一个请求，也不要会刚好读取到一个完整对请求，
                         * 这里其实是需要编码，解码，也就是通信协议  @todo
                         * 这里如何做，大家可以思考一下，后面我们可以体验netty是否如何优雅处理的。
                         */
                        int rc = clientChannel.read(buf);//解析请求完毕
                        //转发请求到具体的业务线程；当然，这里其实可以向dubbo那样，支持转发策略，如果执行时间短，
                        //，比如没有数据库操作等，可以在io线程中执行。本实例，转发到业务线程池
                        dispatch(clientChannel, buf);
                    }
                } catch (Throwable e) {
                    e.printStackTrace();
                    System.out.println("客户端主动断开连接。。。。。。。");
                }
            }
​
            // 处理完事件后，我们还需要处理其他任务，这些任务通常来自业务线程需要IO线程执行的任务
            if (!taskList.isEmpty()) {
                try {
                    taskMainLock.lock();
                    for (Iterator<NioTask> it = taskList
                            .iterator(); it.hasNext();) {
                        NioTask task = it.next();
                        try {
                            SocketChannel sc = task.getSc();
                            if(task.getData() != null ) { // 写操作
                                ByteBuffer byteBuffer = (ByteBuffer)task.getData();
                                byteBuffer.flip();
                                // 如果调用通道的写函数，如果写入的字节数小于0，并且待写入还有剩余空间，说明缓存区已满
                                // 需要注册写事件，等缓存区空闲后继续写入
                                int wc = sc.write(byteBuffer);
                                System.out.println("服务端向客户端发送数据。。。");
                                if(wc < 1 && byteBuffer.hasRemaining()) { // 说明写缓存区已满，需要注册写事件
                                    sc.register(selector, task.getOp(), task.getData());
                                    continue;
                                } 
                                byteBuffer = null;//释放内存
                            } else {
                                sc.register(selector, task.getOp());
                            }
                        } catch (Throwable e) {
                            e.printStackTrace();// ignore
                        }
                        it.remove();
                    }
​
                } finally {
                    taskMainLock.unlock();
                }
            }
        }
    }
}

public class Handler implements Runnable{
​
  // 模拟业务处理
    private static final byte[] b = "hello,服务器收到了你的信息。".getBytes(); // 服务端给客户端的响应
    // 网络通道
    private SocketChannel sc;
    // 请求报文
    private ByteBuffer reqBuffer;
    // IO线程
    private SubReactorThread parent;
​
    public Handler(SocketChannel sc, ByteBuffer reqBuffer,
                   SubReactorThread parent) {
        super();
        this.sc = sc;
        this.reqBuffer = reqBuffer;
        this.parent = parent;
    }
​
    public void run() {
        System.out.println("业务在handler中开始执行。。。");
        // TODO Auto-generated method stub
        //业务处理
        reqBuffer.put(b);
      // 业务处理完成后，通过向IO线程提交任务
        parent.register(new NioTask(sc, SelectionKey.OP_WRITE, reqBuffer));
        System.out.println("业务在handler中执行结束。。。");
    }
}

public class NioClient {
    public static void main(String[] args) {
        // socket
        SocketChannel clientClient;
        // 事件选择器
        Selector selector = null;
        try {
            // 创建网络通道
            clientClient = SocketChannel.open();
            // 设置为非阻塞模型
            clientClient.configureBlocking(false);
            selector = Selector.open();
            // 注册连接成功事件，在与服务端通过tcp三次握手建立连接后可以收到该事件
            clientClient.register(selector, SelectionKey.OP_CONNECT);
          //建立连接，该方法会立即返回
            clientClient.connect(new InetSocketAddress("127.0.0.1",9080));
            Set<SelectionKey> ops = null;
            while(true) {
                try {
                   // 执行事件选择
                    selector.select();
                    ops = selector.selectedKeys();
                    for (Iterator<SelectionKey> it = ops.iterator(); it.hasNext();) {
                        SelectionKey key = it.next();
                        it.remove();
                        if(key.isConnectable())  //连接事件
                            System.out.println("client connect");
                            SocketChannel sc =  (SocketChannel) key.channel();
                            // 判断此通道上是否正在进行连接操作。
                            // 完成套接字通道的连接过程。
                            if (sc.isConnectionPending()) {
                                sc.finishConnect();
                                System.out.println("完成连接!");
                              
                                // 完成连接后，向服务端发送请求包
                                ByteBuffer buffer = ByteBuffer.allocate(1024);
                                buffer.put("Hello,Server".getBytes());
                                buffer.flip();
                                sc.write(buffer);
                            }
                            // 注册读事件，等待服务端响应包到达
                            sc.register(selector, SelectionKey.OP_READ);
                        } else if(key.isWritable()) {
                            System.out.println("客户端写");
                            SocketChannel sc = (SocketChannel)key.channel();
                            //这里是NIO ByteBuffer的基本API
                            ByteBuffer buffer = ByteBuffer.allocate(1024);
                            buffer.put("hello server.".getBytes());
                            buffer.flip();
                            sc.write(buffer);
                        } else if(key.isReadable()) {
                            System.out.println("客户端收到服务器的响应....");
                            SocketChannel sc = (SocketChannel)key.channel();
                            ByteBuffer buffer = ByteBuffer.allocate(1024);
                            int count = sc.read(buffer);
                            if(count > 0 ) {
                                buffer.flip();
                                byte[] response = new byte[buffer.remaining()];
                                buffer.get(response);
                                System.out.println(new String(response));
                            }
                            // 再次发送消息，重复输出
                            buffer = ByteBuffer.allocate(1024);
                            buffer.put("hello server.".getBytes());
                            buffer.flip();
                            sc.write(buffer);
                        }
                    }
                } catch(Throwable e) {
                    e.printStackTrace();
                }
            }
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}