印象中有netty和Redis用了这个模式。既然这么牛逼的中间件高并发框架

介绍

reactor模式是事件驱动的，有一个或多个并发输入源，有一个service Handler，有多个Request Handlers；这个Service Handler会同步的将输入的请求（Event）多路复用的分发给相应的Request Handler。
​​​​​​

Reactor软件工程

下图是reactor的架构设计图

reactor是一种模式，或者说是一种软件工程的概念。 相当于设计模式，是一个概念不是一个实体。 不过JavaNIO的reactor实现真的有一种epoll的感觉，select，selectionKeys

• Synchronous Event Demultiplexer 阻塞等待一系列的Handle中的事件到来，如果阻塞等待返回，即表示在返回的Handle中可以不阻塞的执行返回的事件类型。相当于java中的selector selector.select selectionkeys
• handle 即操作系统中的句柄，是对资源在操作系统层面上的一种抽象，它可以是打开的文件、一个连接(Socket)、Timer等。在Java中相当于一个 channel SocketCannel.open();
• Initiation Dispatcher：用于管理Event Handler，即EventHandler的容器，用以注册、移除EventHandler等；另外，它还作为Reactor模式的入口调用Synchronous Event Demultiplexer的select方法以阻塞等待事件返回，当阻塞等待返回时，根据事件发生的Handle将其分发给对应的Event Handler处理，即回调EventHandler中的handle_event()方法。 SelectionKey
• Event Handler：定义事件处理方法：handle_event()，以供InitiationDispatcher回调使用。 interface
• Concrete Event Handler：事件EventHandler接口，实现特定事件处理逻辑。 interfaceimpl

java代码

https://blog.csdn.net/a953713428/article/details/64907250

服务端

package reactor;import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
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.util.Iterator;
import java.util.Set;/*** @Author zhangyong* @Date 2020/8/6 14:59*/
public class Server {//标识数字/private int flag = 0;//缓冲区大小/private int BLOCK = 4096;//接受数据缓冲区/private ByteBuffer sendbuffer = ByteBuffer.allocate(BLOCK);//发送数据缓冲区/private ByteBuffer receivebuffer = ByteBuffer.allocate(BLOCK);private Selector selector;public static void main(String[] args) throws IOException {// TODO Auto-generated method stubint port = 7788;Server server = new Server(port);server.listen();}public Server(int port) throws IOException {// 打开服务器套接字通道ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();// 服务器配置为非阻塞serverSocketChannel.configureBlocking(false);// 检索与此通道关联的服务器套接字ServerSocket serverSocket = serverSocketChannel.socket();// 进行服务的绑定serverSocket.bind(new InetSocketAddress(port));// 通过open()方法找到Selectorselector = Selector.open();// 注册到selector，等待连接serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);System.out.println("Server Start----7788:");}// 监听private void listen() throws IOException {while (true) {// 选择一组键，并且相应的通道已经打开selector.select();// 返回此选择器的已选择键集。Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()) {SelectionKey selectionKey = iterator.next();iterator.remove();handleKey(selectionKey);}}}// 处理请求private void handleKey(SelectionKey selectionKey) throws IOException {// 接受请求ServerSocketChannel server = null;SocketChannel client = null;String receiveText;String sendText;int count = 0;// 测试此键的通道是否已准备好接受新的套接字连接。if (selectionKey.isAcceptable()) {// 返回为之创建此键的通道。server = (ServerSocketChannel) selectionKey.channel();// 接受到此通道套接字的连接。// 此方法返回的套接字通道（如果有）将处于阻塞模式。  接受请求client = server.accept();// 配置为非阻塞client.configureBlocking(false);// 注册到selector，等待连接client.register(selector, SelectionKey.OP_READ);  //将事件监听读操作，下次就绪时进行读取。} else if (selectionKey.isReadable()) {// 返回为之创建此键的通道。client = (SocketChannel) selectionKey.channel();//将缓冲区清空以备下次读取receivebuffer.clear();//读取服务器发送来的数据到缓冲区中count = client.read(receivebuffer);if (count > 0) {receiveText = new String(receivebuffer.array(), 0, count);System.out.println("服务器端接受客户端数据--:" + receiveText);client.register(selector, SelectionKey.OP_WRITE);       //读完数据，准备返回信息，将channel注册为写事件。供返回数据。}} else if (selectionKey.isWritable()) {//将缓冲区清空以备下次写入sendbuffer.clear();// 返回为之创建此键的通道。client = (SocketChannel) selectionKey.channel();sendText = "message from server--" + flag++;//向缓冲区中输入数据sendbuffer.put(sendText.getBytes());//将缓冲区各标志复位,因为向里面put了数据标志被改变要想从中读取数据发向服务器,就要复位sendbuffer.flip();//输出到通道client.write(sendbuffer);System.out.println("服务器端向客户端发送数据--：" + sendText);client.register(selector, SelectionKey.OP_READ);}}
}

客户端

package reactor;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.Set;/*** @Author zhangyong* @Date 2020/8/6 15:00*/
public class Client {//标识数字/private static int flag = 0;//缓冲区大小/private static int BLOCK = 4096;//接受数据缓冲区/private static ByteBuffer sendbuffer = ByteBuffer.allocate(BLOCK);//发送数据缓冲区/private static ByteBuffer receivebuffer = ByteBuffer.allocate(BLOCK);//服务器端地址/private final static InetSocketAddress SERVER_ADDRESS = new InetSocketAddress("localhost", 7788);public static void main(String[] args) throws IOException {// TODO Auto-generated method stub// 打开socket通道SocketChannel socketChannel = SocketChannel.open();// 设置为非阻塞方式socketChannel.configureBlocking(false);// 打开选择器Selector selector = Selector.open();// 注册连接服务端socket动作socketChannel.register(selector, SelectionKey.OP_CONNECT);// 连接socketChannel.connect(SERVER_ADDRESS);// 分配缓冲区大小内存Set<SelectionKey> selectionKeys;Iterator<SelectionKey> iterator;SelectionKey selectionKey;SocketChannel client;String receiveText;String sendText;int count = 0;while (true) {//选择一组键，其相应的通道已为 I/O 操作准备就绪。//此方法执行处于阻塞模式的选择操作。selector.select();//返回此选择器的已选择键集。selectionKeys = selector.selectedKeys();//System.out.println(selectionKeys.size());iterator = selectionKeys.iterator();while (iterator.hasNext()) {selectionKey = iterator.next();if (selectionKey.isConnectable()) {System.out.println("client connect");client = (SocketChannel) selectionKey.channel();// 判断此通道上是否正在进行连接操作。// 完成套接字通道的连接过程。if (client.isConnectionPending()) {client.finishConnect();System.out.println("完成连接!");sendbuffer.clear();sendbuffer.put("Hello,Server".getBytes());sendbuffer.flip();client.write(sendbuffer);}client.register(selector, SelectionKey.OP_READ);} else if (selectionKey.isReadable()) {client = (SocketChannel) selectionKey.channel();//将缓冲区清空以备下次读取receivebuffer.clear();//读取服务器发送来的数据到缓冲区中count = client.read(receivebuffer);if (count > 0) {receiveText = new String(receivebuffer.array(), 0, count);System.out.println("客户端接受服务器端数据--:" + receiveText);client.register(selector, SelectionKey.OP_WRITE);}} else if (selectionKey.isWritable()) {sendbuffer.clear();client = (SocketChannel) selectionKey.channel();sendText = "message from client--" + (flag++);sendbuffer.put(sendText.getBytes());//将缓冲区各标志复位,因为向里面put了数据标志被改变要想从中读取数据发向服务器,就要复位sendbuffer.flip();client.write(sendbuffer);System.out.println("客户端向服务器端发送数据--：" + sendText);client.register(selector, SelectionKey.OP_READ);}}selectionKeys.clear();}}
}

总结

reactor是一种设计模式，并不是一个实体。基于事件来处理，一个线程负责监听，其他线程负责处理监听到事件时候的handle