经典Reactor反应器模式_vlambda技术博客

vlambda
2022-04-26

经典Reactor反应器模式

反应器模式是一种为处理服务请求并发提交到一个或者多个服务处理程序的事件设计模式。

1. Reactor三步曲

（1）注册

将就绪事件，注册到选择器上。channel->selector

（2）轮询

选择器进行轮询，其上面的事件是否就绪。selector->channel

（3）分发

将就绪的事件，分发到对应的处理器进行处理。selector->handler

2. 实现Reactor反应器模式

以JAVA NIO为基础，实现一个基于反应器模式的事件处理模型。如：当有新连接到达时，处理新连接。

创建反应器类ServerReactor，一个反应器对应一个选择器Selector轮询事件。

第一步：注册

将通道Channel的accept事件注册到Selecotor上，返回选择键 SelectionKey

public class ServerReactor implements Runnable{
 //选择器 Selector selector; //通道 ServerSocketChannel socketChannel;
   //初始化,将通道的accept事件注册到selector上    public ServerReactor() throws IOException { //获取selector和channel selector = Selector.open(); socketChannel = ServerSocketChannel.open(); //处理socket连接 ServerSocket socket = socketChannel.socket(); //绑定监听的端口号 InetSocketAddress address = new InetSocketAddress(Const.port); socket.bind(address); //设置为非阻塞 socketChannel.configureBlocking(false); //1.注册accept事件 SelectionKey sk = socketChannel.register(selector,SelectionKey.OP_ACCEPT); //将新连接处理器作为附件，绑定到sk选择键上 sk.attach(new AcceptHandler()); } }

获取Nio的Channel和Selecotor，获取通道对应的socket，并绑定监听的网络端口，将通道Channel的accept事件注册到Selecotor上，返回选择键 SelectionKey ，并将处理accept事件的处理器AcceptHandler绑定到返回的选择键 SelectionKey 上。

第二步：轮询

反应器线程对应的需要轮询其Selecotor上面的就绪通道事件

public class ServerReactor implements Runnable{
 //选择器 Selector selector; //通道 ServerSocketChannel socketChannel;
     .....      @Override public void run() { //2.选择器轮询 try { while (!Thread.interrupted()){ selector.select();                //获取选择器上的就绪选择键 Set<SelectionKey> selected =selector.selectedKeys(); Iterator<SelectionKey> it = selected.iterator(); while (it.hasNext()){ //3.反应器负责dispatch收到的事件 SelectionKey sk = it.next(); dispatch(sk); } selected.clear(); } }catch (Exception e){ e.printStackTrace(); } }}

第三步：分发

将轮询到的事件分发到对应的Handler事件处理器进行处理

public class ServerReactor implements Runnable{
 //选择器 Selector selector; //通道 ServerSocketChannel socketChannel;
     .....  @Override public void run() {       .....      //3.反应器负责dispatch收到的事件 SelectionKey sk = it.next(); dispatch(sk); }  //反应器的分发 void dispatch(SelectionKey sk){ Runnable handler = (Runnable) sk.attachment();        //调用之间绑定到选择键的handler处理器对象（对应各种事件handler） if (handler!=null){ handler.run(); }    } }
//Handler：对应的连接处理器class AcceptHandler implements Runnable{ @Override public void run() { try { SocketChannel channel = socketChannel.accept(); if (channel!=null){                    //连接好后，交给另外的处理器进行处理 new EchoHandler(selector,channel); } }catch (Exception e){ e.printStackTrace(); } }}