reactor

什么是Reactor模式

Reactor 模式是一種事件驅動架構的實現技術

The reactor design pattern is an event handling pattern for handling service requests delivered concurrently to a service handler by one or more inputs. The service handler then demultiplexes the incoming requests and dispatches them synchronously to the associated request handlers.

我們知道Reactor模式首先是事件驅動的，有一個或多個并發輸入源，有一個Service Handler，有多個Request Handlers；這個Service Handler會同步的將輸入的請求(Event)多路復用的分發給相應的 Request Handler。

標準reactor模式

Handle

Handle代表操作系統管理的資源，包括：網絡鏈接，打開的文件，計時器，同步對象等等。在我們的web系統中，Handle代表與客戶端連接的套接字，Synchronous Event Demultiplexer在這些套接字上等待事件的發生。

Synchronous Event Demultiplexer

在一個Handle集合上等待事件的發生。這里常用系統調用select[1]，UNIX和WIN32平臺都支持這個系統調用。select的返回結果說明handle上發生情況，需要被處理。

Initiation Dispatcher

提供接口：注冊，刪除和派發Event Handler。上面的Synchronous Event Demultiplexer等待事件的發生，當檢測到新的事件，就把事件交給Initiation Dispatcher，它去回調Event Handler。事件種類一般有：接受到連接，數據輸入，數據輸出，超時。

Event Handler

定義一個抽象接口，包含一個鉤子方法，實現特定服務的派發操作。這個方法實現了與特定應用相關的服務。

Concrete Event Handler

繼承上面的類，實現鉤子方法。應用把Concrete Event Handler注冊到Initiation Dispatcher，等待被處理的事件。當事件發生，這些方法被回調。

Reactor pattern in Java NIO

我們將會基于java nio I/O 復用模型 實現reactor模式的demo

Java NIO 是為了彌補傳統 I/O 工作模式的不足而研發的，NIO 的工具包提出了基于 Selector(選擇器)、Buffer(緩沖區)、Channel(通道)的新模式；Selector、Channel和 SelectionKey(選擇鍵)配合起來使用，可以實現并發的非阻塞型 I/O 能力。

java reactor實現

讓我們將java reactor的實現與標準意義對應起來：

Selector (demultiplexer)

Selector is the Java building block, which is analogous to the demultiplexer in the Reactor pattern. Selector is where you register your interest in various I/O events and the objects tell you when those events occur.

Reactor/initiation dispatcher

We should use the Java NIO Selector in the Dispatcher/Reactor. For this, we can introduce our own Dispatcher/Reactor implementation called ‘Reactor’. The reactor comprises java.nio.channels.Selector and a map of registered handlers. As per the definition of the Dispatcher/Reactor, ‘Reactor’ will call the Selector.select() while waiting for the IO event to occur.

Handle

對應 SelectionKey.

Event

不同的IO events 如 SlectionKey.OP_READ ， SelectionKey.OP_ACCEPT，SelectionKey.OP_WRITE，SelectionKey.OP_CONNECT

Handler

事件處理器，A handler is often implemented as runnable or callable in Java.

code time

https://github.com/kasun04/rnd/tree/master/nio-reactor

在我們的代碼里通過一個map來管理我們關心的事件和對應的事件handler。

當我們通過Selector.select來輪詢到來的事件。并遍歷Set.

while (true) { // Loop indefinitely

demultiplexer.select();// 是阻塞的

Set readyHandles =

demultiplexer.selectedKeys();

Iterator handleIterator =

readyHandles.iterator();

while (handleIterator.hasNext()) {

SelectionKey handle = handleIterator.next();

if (handle.isAcceptable()) {

EventHandler handler =

registeredHandlers.get(SelectionKey.OP_ACCEPT);

handler.handleEvent(handle);

// 需要將此次事件key移除，避免重復處理

handleIterator.remove();

}

if (handle.isReadable()) {

EventHandler handler =

registeredHandlers.get(SelectionKey.OP_READ);

handler.handleEvent(handle);

handleIterator.remove();

}

if (handle.isWritable()) {

EventHandler handler =

registeredHandlers.get(SelectionKey.OP_WRITE);

handler.handleEvent(handle);

handleIterator.remove();

}

}

}

我們在handleEvent里處理事件，并且注冊新的關心的事件，比如在AcceptEventHandler處理完后,注冊SelectionKey.OP_READ，將事件拋給下一個handler。并可以在

public void handleEvent(SelectionKey handle) throws Exception {

System.out.println("===== Accept Event Handler =====");

ServerSocketChannel serverSocketChannel =

(ServerSocketChannel) handle.channel();

SocketChannel socketChannel = serverSocketChannel.accept();

if (socketChannel != null) {

socketChannel.configureBlocking(false);

socketChannel.register(

demultiplexer, SelectionKey.OP_READ);

}

}

我們看到上面輪詢的代碼其實是單線程的每一個handler的handleEvent都是在Reactor這個一個里的，為什么不做成多線程呢？

NIO由原來的BIO的阻塞讀寫(占用線程)變成了單線程輪詢事件，找到可以進行讀寫的網絡描述符進行讀寫。除了事件的輪詢是阻塞的(沒有可干的事情必須要阻塞)，剩余的I/O操作都是純CPU操作，沒有必要開啟多線程。

我們看read事件的handler,為了提高效率，業務代碼開線程，將io處理與業務代碼處理分離。

public void handleEvent(SelectionKey handle) throws Exception {

System.out.println("===== Read Event Handler =====");

SocketChannel socketChannel =

(SocketChannel) handle.channel();

// 從socket讀取數據

byte[] buffer=read(socketChannel);

// 此處執行業務操作，最好單線程或者多線程，將io處理與業務代碼處理分離

doBusiness(buffer);

// Rewind the buffer to start reading from the beginning

// Register the interest for writable readiness event for

// this channel in order to echo back the message

socketChannel.register(

demultiplexer, SelectionKey.OP_WRITE, inputBuffer);

}

BIO與NIO的對比

美團技術博客--Java NIO淺析

關鍵差別:

BIO accept,read,write 都是阻塞的。

NIO select 阻塞，read，write是非阻塞的。

bio 經典模型

{

ExecutorService executor = Excutors.newFixedThreadPollExecutor(100);//線程池

ServerSocket serverSocket = new ServerSocket();

serverSocket.bind(8088);

while(!Thread.currentThread.isInturrupted()){//主線程死循環等待新連接到來

Socket socket = serverSocket.accept();//阻塞

//為新的連接創建新的線程

executor.submit(new ConnectIOnHandler(socket));

}

class ConnectIOnHandler extends Thread{

private Socket socket;

public ConnectIOnHandler(Socket socket){

this.socket = socket;

}

public void run(){

while(!Thread.currentThread.isInturrupted()&&!socket.isClosed()){死循環處理讀寫事件

String someThing = socket.read()....//讀取數據

if(someThing!=null){

......//處理數據

socket.write()....//寫數據

}

}

}

}

這是一個經典的每連接每線程的模型，之所以使用多線程，主要原因在于socket.accept()、socket.read()、socket.write()三個主要函數都是同步阻塞的，當一個連接在處理I/O的時候，系統是阻塞的，如果是單線程的話必然就掛死在那里(比如阻塞在read，那么就無法accept到其他請求了)；但CPU是被釋放出來的，開啟多線程，就可以讓CPU去處理更多的事情。

NIO reactor模型

如上設計和分析:

由于read，write的非阻塞，所以可以不用多線程，并且由于線程的節約，連接數大的時候因為線程切換帶來的問題也隨之解決，進而為處理海量連接提供了可能。

單線程處理I/O的效率確實非常高，沒有線程切換，只是拼命的讀、寫、選擇事件。但現在的服務器，一般都是多核處理器，如果能夠利用多核心進行I/O，無疑對效率會有更大的提高。

限制:

Java的Selector對于Linux系統來說，有一個致命限制：同一個channel的select不能被并發的調用。因此，如果有多個I/O線程，必須保證：一個socket只能屬于一個IoThread，而一個IoThread可以管理多個socket。

總結

以上是生活随笔為你收集整理的java reactor nio_java reactor与NIO的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。