在Reactor基本线程模型中，Doug Lea将NIO进行accept操作的部分提取出来，通过一个单一线程acceptor（也就是当前线程）实现client的accept信号的监听，并进行分发（进行后续事件的注册）。
而当监听到read等事件后，通过dispatch将相关事件处理分发到线程池TreadPool中，交由worker thread进行具体业务处理。
当然这样的线程模型，其扩展性依旧无法满足需求，其性能瓶颈，会卡在acceptor线程上。所以Doug Lea进而提出了multiple Reactors
其设计是将原先的基本Reactor线程模型的Reactor拆分为mainReactor与subReactor，中间通过acceptor连接，从而降低原先基本Reactor线程模型中acceptor的压力。
优点优秀的可扩展性
更高的性能瓶颈
缺点需要NIO的理解基础
需要理解Reactor线程模型
代码实现较为复杂（相较原先的NIO与BIO）
代码示例这里给出一些简单的demo，供大家认识。
NIO_Client（和之前的NIO_Client没有任何区别） package tech.jarry.learning.netease; import java.io.IOException; import java.net.Inet4Address; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SocketChannel; import java.util.Scanner; /** * @Description： NIO模型下的TCP客户端实现 * @Author: jarry */ public class NIOClient { public static void main(String[] args) throws IOException { // 获得一个SocektChannel SocketChannel socketChannel = SocketChannel.open(); // 设置SocketChannel为非阻塞模式 socketChannel.configureBlocking(false); // 设置SocketChannel的连接配置 socketChannel.connect(new InetSocketAddress(Inet4Address.getLocalHost(), 8080)); // 通过循环，不断连接。跳出循环，表示连接建立成功 while (!socketChannel.finishConnect()){ // 如果没有成功建立连接，就一直阻塞当前线程（.yield()会令当前线程“谦让”出CPU资源） Thread.yield(); } // 发送外部输入的数据 Scanner scanner = new Scanner(System.in); System.out.println("please input:"); String msg = scanner.nextLine(); // ByteBuffer.wrap()会直接调用HeapByteBuffer。故一方面其会自己完成内存分配。另一方面，其分配的内存是非直接内存（非heap堆） ByteBuffer byteBuffer = ByteBuffer.wrap(msg.getBytes()); // ByteBuffer.hasRemaining()用于判断对应ByteBuffer是否还有剩余数据（实现：return position < limit;） while (byteBuffer.hasRemaining()){ socketChannel.write(byteBuffer); } // 读取响应 System.out.println("receive echoResponse from server"); // 设置缓冲区大小为1024 ByteBuffer requestBuffer = ByteBuffer.allocate(1024); // 判断条件：是否开启，是否读取到数据 //TODO 我认为这里的实现十分粗糙，是不可以置于生产环境的，具体还需要后面再看看（即使是过渡demo，也可以思考一下嘛） while (socketChannel.isOpen() && socketChannel.read(requestBuffer) != -1){ // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了) if (requestBuffer.position() > 0) { break; } } requestBuffer.flip(); // byte[] content = new byte[requestBuffer.limit()]; // // .get()方法只会返回byte类型（猜测是当前标记位的数据） // requestBuffer.get(content); // System.out.println(new String(content)); // ByteBuffer提供了大量的基本类型转换的方法，可以直接拿来使用 System.out.println(new String(requestBuffer.array())); scanner.close(); socketChannel.close(); } } NIO_ServerV4_ReactorV1 package tech.jarry.learning.netease; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.*; import java.util.Iterator; import java.util.Random; import java.util.Set; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; /** * @Description： 根据Doug Lea大神的多路复用Reactor线程模型，进行编码，学习Reactor设计模式在网络编程的重要体现 * 注意：NIOServerV2作为一个demo已经不错了。但是仍然存在致命的性能瓶颈（其实很明显，整个网络编程就靠一个线程实现全部工作，肯定不行，起码没法充分发挥多核CPU的能力） * 故将服务端常用的部分分为accept,read,bussinessDeal三个部分（第三部分，本demo就不深入了） * @Author: jarry */ public class NIOServerV3 { // 处理业务操作的线程 private static ExecutorService workPool = Executors.newCachedThreadPool(); /** * 封装了Selector.select()等事件的轮询的共用代码 */ abstract class ReactorThread extends Thread { Selector selector; LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>(); /** * Selector监听到有事件后,调用这个方法（不过具体实现，需要基类自己实现） * @param channel */ public abstract void handler(SelectableChannel channel) throws Exception; private ReactorThread() throws IOException { selector = Selector.open(); } // 用于判断线程运行状态 volatile boolean running = false; @Override public void run() { // 轮询Selector事件 while (running) { try { // 执行队列中的任务 Runnable task; while ((task = taskQueue.poll()) != null) { task.run(); } selector.select(1000); // 获取查询结果 Set<SelectionKey> selected = selector.selectedKeys(); // 遍历查询结果 Iterator<SelectionKey> iter = selected.iterator(); while (iter.hasNext()) { // 被封装的查询结果 SelectionKey key = iter.next(); iter.remove(); int readyOps = key.readyOps(); // 关注 Read 和 Accept两个事件 if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { try { SelectableChannel channel = (SelectableChannel) key.attachment(); channel.configureBlocking(false); handler(channel); if (!channel.isOpen()) { key.cancel(); // 如果关闭了,就取消这个KEY的订阅 } } catch (Exception ex) { key.cancel(); // 如果有异常,就取消这个KEY的订阅 } } } selector.selectNow(); } catch (IOException e) { e.printStackTrace(); } } } private SelectionKey register(SelectableChannel channel) throws Exception { // 为什么register要以任务提交的形式，让reactor线程去处理？ // 因为线程在执行channel注册到selector的过程中，会和调用selector.select()方法的线程争用同一把锁 // 而select()方法是在eventLoop中通过while循环调用的，争抢的可能性很高，为了让register能更快的执行，就放到同一个线程来处理 FutureTask<SelectionKey> futureTask = new FutureTask<>(() -> channel.register(selector, 0, channel)); taskQueue.add(futureTask); return futureTask.get(); } private void doStart() { if (!running) { running = true; start(); } } } // 0. 创建ServerSocketChannel private ServerSocketChannel serverSocketChannel; // 1、创建多个线程 - accept处理reactor线程 (accept线程) private ReactorThread[] mainReactorThreads = new ReactorThread[1]; // 2、创建多个线程 - io处理reactor线程 (I/O线程) private ReactorThread[] subReactorThreads = new ReactorThread[8]; /** * 初始化线程组 */ private void initGroup() throws IOException { // 创建IO线程,负责处理客户端连接以后socketChannel的IO读写 for (int i = 0; i < subReactorThreads.length; i++) { subReactorThreads[i] = new ReactorThread() { @Override public void handler(SelectableChannel channel) throws IOException { // work线程只负责处理IO处理，不处理accept事件 SocketChannel ch = (SocketChannel) channel; ByteBuffer requestBuffer = ByteBuffer.allocate(1024); while (ch.isOpen() && ch.read(requestBuffer) != -1) { // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了) if (requestBuffer.position() > 0) { break; } } if (requestBuffer.position() == 0) { return; // 如果没数据了, 则不继续后面的处理 } requestBuffer.flip(); byte[] content = new byte[requestBuffer.limit()]; requestBuffer.get(content); System.out.println(new String(content)); System.out.println(Thread.currentThread().getName() + "收到数据,来自：" + ch.getRemoteAddress()); // TODO 业务操作 数据库、接口... workPool.submit(() -> { }); // 响应结果 200 String response = "HTTP/1.1 200 OK\r\n" + "Content-Length: 11\r\n\r\n" + "Hello World"; ByteBuffer buffer = ByteBuffer.wrap(response.getBytes()); while (buffer.hasRemaining()) { ch.write(buffer); } } }; } // 创建mainReactor线程, 只负责处理serverSocketChannel for (int i = 0; i < mainReactorThreads.length; i++) { mainReactorThreads[i] = new ReactorThread() { AtomicInteger incr = new AtomicInteger(0); @Override public void handler(SelectableChannel channel) throws Exception { // 只做请求分发，不做具体的数据读取 ServerSocketChannel ch = (ServerSocketChannel) channel; SocketChannel socketChannel = ch.accept(); socketChannel.configureBlocking(false); // 收到连接建立的通知之后，分发给I/O线程继续去读取数据 int index = incr.getAndIncrement() % subReactorThreads.length; ReactorThread workEventLoop = subReactorThreads[index]; workEventLoop.doStart(); SelectionKey selectionKey = workEventLoop.register(socketChannel); selectionKey.interestOps(SelectionKey.OP_READ); System.out.println(Thread.currentThread().getName() + "收到新连接 : " + socketChannel.getRemoteAddress()); } }; } } /** * 初始化channel,并且绑定一个eventLoop线程 * * @throws IOException IO异常 */ private void initAndRegister() throws Exception { // 1、 创建ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.configureBlocking(false); // 2、 将serverSocketChannel注册到selector int index = new Random().nextInt(mainReactorThreads.length); mainReactorThreads[index].doStart(); SelectionKey selectionKey = mainReactorThreads[index].register(serverSocketChannel); selectionKey.interestOps(SelectionKey.OP_ACCEPT); } /** * 绑定端口 * * @throws IOException IO异常 */ private void bind() throws IOException { // 1、 正式绑定端口，对外服务 serverSocketChannel.bind(new InetSocketAddress(8080)); System.out.println("启动完成，端口8080"); } public static void main(String[] args) throws Exception { NIOServerV3 nioServerV3 = new NIOServerV3(); nioServerV3.initGroup(); // 1、 创建main和sub两组线程 nioServerV3.initAndRegister(); // 2、 创建serverSocketChannel，注册到mainReactor线程上的selector上 nioServerV3.bind(); // 3、 为serverSocketChannel绑定端口 } } NIO_ServerV4_ReactorV2为了更好的理解Reactor线程模型，我将之前的Reactor代码，按照我的代码习惯，做了一些调整。