标签：

Reactor 模式简单实现

在网上有部分文章在描述Netty时，会提到Reactor。这个Reactor到底是什么呢？为了搞清楚Reactor到底是什么鬼，我写了一个简单的Demo，来帮助大家理解他。

网上是这么描述Reactor的：

The Reactor design pattern handles service requests that are delivered concurrently to an application by one or more clients.

Each service in an application may consist of serveral methods and is represented by a separate event handler that is responsible for dispatching service-specific requests.

Dispatching of event handlers is performed by an initiation dispatcher, which manages the registered event handlers. Demultiplexing of service requests is performed by a synchronous event demultiplexer.

大致意思是：Reactor是用于处理多个客户端的请求的设计模式。应用程序提供的每一种服务都可能包括多个方法，并且有必要为这每一个服务分配独立的请求处理器(也可以说是 event handler)。对于Event handler的调度是有Dispatcher来执行的，这个Dispatcher可以管理event handler的注册工作。而分离器Demultiplexer则将一个服务分成了多份。这段话看起来还是不那么容易理解的。

我对这段话的理解是：应用程序提供多种服务，而每一种服务都会分为多步骤(或者多类别)进行。这里将每一步都作为一个事件，那么每一步的处理就认为是一个event handler。Dispatcher管理这多个步骤的处理器，也即dispatcher管理着多个Event Handler。而将一个服务处理分为多步骤(多个类别)的处理的工作则是有分离器来完成。

从这个类图上看，主要有四个角色：

·Handle：事件源。

·EventHandler：事件处理器

·Dispatcher：调度器。使用Demultiplexer选出可以执行处理的EventHandler，然后执行对EventHandler的调度。

·Demultiplexer：同步的事件分离器。

从类图上看：Dispatcher中有一个Selector和一个EventHandler集合(可以是List，也可以是Map，具体怎么实现根据实际需求)。Regist_handler、remove_handler用于管理EventHandler。

Handle_events用于启动调度，这个方法的实现通常是：使用分离器选择出可以调度的Event，然后对它们进行调度。

下面就是对Reactor模式的简单实现：

packagecom.fjn.jdk.nio.reactor.standard;importjava.util.ArrayList;importjava.util.Date;importjava.util.List;importjava.util.Map;importjava.util.concurrent.BlockingQueue;importjava.util.concurrent.ConcurrentHashMap;importjava.util.concurrent.LinkedBlockingQueue;public classStandardReactor {

}classEventDispatcher {

Map eventHandlerMap = new ConcurrentHashMap();

Demultiplexer selector;

EventDispatcher(Demultiplexer selector) {this.selector =selector;

}public voidregistEventHandler(EventType eventType, EventHandler eventHandler) {

eventHandlerMap.put(eventType, eventHandler);

}public voidremoveEventHandler(EventType eventType) {

eventHandlerMap.remove(eventType);

}public voidhandleEvents() {

dispatch();

}private voiddispatch() {while (true) {

List events =selector.select();for(Event event : events) {

EventHandler eventHandler=eventHandlerMap.get(event.type);

eventHandler.handle(event);

}

}

}

}classDemultiplexer {private BlockingQueue eventQueue = new LinkedBlockingQueue();private Object lock = newObject();

Listselect() {return select(0);

}

List select(longtimeout) {if (timeout > 0) {if(eventQueue.isEmpty()) {synchronized(lock) {if(eventQueue.isEmpty()) {try{

lock.wait(timeout);

}catch(InterruptedException e) {//ignore it

}

}

}

}

}

List events = new ArrayList();

eventQueue.drainTo(events);returnevents;

}public voidaddEvent(Event e) {boolean success =eventQueue.offer(e);if(success) {synchronized(lock) {

lock.notify();

}

}

}

}classSource {private Date date = newDate();private String id = date.toString() + "_" +System.identityHashCode(date);

@OverridepublicString toString() {returnid;

}

}enumEventType {

ACCEPT, READ, WRITE, TIMEOUT;

}classEvent {publicEventType type;publicSource source;

}abstract classEventHandler {

Source source;public abstract voidhandle(Event event);publicSource getSource() {returnsource;

}

}class AcceptEventHandler extendsEventHandler {privateDemultiplexer selector;publicAcceptEventHandler(Demultiplexer selector) {this.selector =selector;

}

@Overridepublic voidhandle(Event event) {if (event.type ==EventType.ACCEPT) {

Event readEvent= newEvent();

readEvent.source=event.source;

readEvent.type=EventType.READ;

selector.addEvent(readEvent);

}

}

}class ReadEventHandler extendsEventHandler {//private Pipeline pipeline;

@Overridepublic voidhandle(Event event) {//create channel with a pipeline//register the channel to this event dispatcher or a child event dispatcher//handle event use the pipeline ://step 1: read to a frame buffer//step 2: use frame decoder to decode buffer as a message (maybe a business object)//step 3: handle the message or submit the message to business thread pool//step 4: register a message event

}

}class WriteEventHandler extendsEventHandler {

@Overridepublic voidhandle(Event event) {//step 1: encode a message to byte[]//step 2: submit a write task to IOWorker thread pool

}

}//-------------------------------分割线--------------------------//

class Acceptor implementsRunnable {private int port; //server socket port

privateDemultiplexer selector;//代表 serversocket

private BlockingQueue sourceQueue = new LinkedBlockingQueue();

Acceptor(Demultiplexer selector,intport) {this.selector =selector;this.port =port;

}public voidaNewConnection(Source source) {

sourceQueue.offer(source);

}public intgetPort() {return this.port;

}public voidrun() {while (true) {

Source source= null;try{//相当于 serversocket.accept()

source =sourceQueue.take();

}catch(InterruptedException e) {//ignore it;

}if (source != null) {

Event acceptEvent= newEvent();

acceptEvent.source=source;

acceptEvent.type=EventType.ACCEPT;

selector.addEvent(acceptEvent);

}

}

}

}classServer {

Demultiplexer selector= newDemultiplexer();

EventDispatcher eventLooper= newEventDispatcher(selector);

Acceptor acceptor;

Server(intport) {

acceptor= newAcceptor(selector, port);

}public voidstart() {

eventLooper.registEventHandler(EventType.ACCEPT,newAcceptEventHandler(selector));new Thread(acceptor, "Acceptor-" +acceptor.getPort()).start();

eventLooper.handleEvents();

}

}

标签：