Reactor 模式的简单实现

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模式的简单实现:

package com.fjn.jdk.nio.reactor.standard;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

public class StandardReactor {

}

class EventDispatcher {
    Map<EventType, EventHandler> eventHandlerMap = new ConcurrentHashMap<EventType, EventHandler>();

    Demultiplexer selector;

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

    public void registEventHandler(EventType eventType, EventHandler eventHandler) {
        eventHandlerMap.put(eventType, eventHandler);

    }

    public void removeEventHandler(EventType eventType) {
        eventHandlerMap.remove(eventType);
    }

    public void handleEvents() {
        dispatch();
    }

    private void dispatch() {
        while (true) {
            List<Event> events = selector.select();

            for (Event event : events) {
                EventHandler eventHandler = eventHandlerMap.get(event.type);
                eventHandler.handle(event);
            }
        }
    }
}

class Demultiplexer {
    private BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<Event>();
    private Object lock = new Object();

    List<Event> select() {
        return select(0);
    }

    List<Event> select(long timeout) {
        if (timeout > 0) {
            if (eventQueue.isEmpty()) {
                synchronized (lock) {
                    if (eventQueue.isEmpty()) {
                        try {
                            lock.wait(timeout);
                        } catch (InterruptedException e) {
                            // ignore it
                        }
                    }
                }

            }
        }
        List<Event> events = new ArrayList<Event>();
        eventQueue.drainTo(events);
        return events;
    }

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

        }
    }

}

class Source {
    private Date date = new Date();
    private String id = date.toString() + "_" + System.identityHashCode(date);

    @Override
    public String toString() {
        return id;
    }
}

enum EventType {
    ACCEPT, READ, WRITE, TIMEOUT;
}

class Event {
    public EventType type;
    public Source source;
}

abstract class EventHandler {
    Source source;

    public abstract void handle(Event event);

    public Source getSource() {
        return source;
    }
}

class AcceptEventHandler extends EventHandler {

    private Demultiplexer selector;

    public AcceptEventHandler(Demultiplexer selector) {
        this.selector = selector;
    }

    @Override
    public void handle(Event event) {
        if (event.type == EventType.ACCEPT) {

            Event readEvent = new Event();
            readEvent.source = event.source;
            readEvent.type = EventType.READ;

            selector.addEvent(readEvent);
        }
    }

}

class ReadEventHandler extends EventHandler {
    // private Pipeline pipeline;

    @Override
    public void handle(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 extends EventHandler {

    @Override
    public void handle(Event event) {
        // step 1: encode a message to byte[]
        // step 2: submit a write task to IOWorker thread pool
    }

}

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

class Acceptor implements Runnable {
    private int port; // server socket port
    private Demultiplexer selector;

    // 代表 serversocket
    private BlockingQueue<Source> sourceQueue = new LinkedBlockingQueue<Source>();

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

    public void aNewConnection(Source source) {
        sourceQueue.offer(source);
    }

    public int getPort() {
        return this.port;
    }

    public void run() {
        while (true) {

            Source source = null;
            try {
                // 相当于 serversocket.accept()
                source = sourceQueue.take();
            } catch (InterruptedException e) {
                // ignore it;
            }

            if (source != null) {
                Event acceptEvent = new Event();
                acceptEvent.source = source;
                acceptEvent.type = EventType.ACCEPT;

                selector.addEvent(acceptEvent);
            }

        }
    }

}

class Server {
    Demultiplexer selector = new Demultiplexer();
    EventDispatcher eventLooper = new EventDispatcher(selector);
    Acceptor acceptor;

    Server(int port) {
        acceptor = new Acceptor(selector, port);
    }

    public void start() {
        eventLooper.registEventHandler(EventType.ACCEPT, new AcceptEventHandler(selector));
        new Thread(acceptor, "Acceptor-" + acceptor.getPort()).start();
        eventLooper.handleEvents();
    }

}
时间: 2024-10-19 11:18:28

Reactor 模式的简单实现的相关文章

【转】Netty那点事(四)Netty与Reactor模式

[原文]https://github.com/code4craft/netty-learning/blob/master/posts/ch4-reactor.md 一:Netty.NIO.多线程? 时隔很久终于又更新了!之前一直迟迟未动也是因为积累不够,后面比较难下手.过年期间@李林锋hw发布了一个Netty5.0架构剖析和源码解读 http://vdisk.weibo.com/s/C9LV9iVqH13rW/1391437855,看完也是收获不少.前面的文章我们分析了Netty的结构,这次咱们

对于观察者模式,Reactor模式,Proactor模式的一点理解

最近就服务器程序IO效率这一块了解一下设计模式中的Reacotr模式和proactor模式,感觉跟观察者模式有些类似的地方,网上也看了一些其他人对三者之间区别的理解,都讲得很仔细,在此根据自己的理解做一点简单的记录和总结,如果理解不对的地方,以后再慢慢深入和更新. 观察者模式: 也可以称为为 发布-订阅 模式,主要适用于多个对象依赖某一个对象的状态并,当某对象状态发生改变时,要通知其他依赖对象做出更新.是一种1对多的关系.当然,如果依赖的对象只有一个时也是一种特殊的一对一关系.通常,观察者模式适

libevent之Reactor模式

通过前边的一篇博文轻量级网络库libevent初探,我们知道libevent实际上是封装了不同操作系统下的/dev/poll.kqueue.event ports.select.poll和epoll事件机制,从而给我们提供一个统一的接口. libevent采用了Reactor I/O 设计模式,而Reactor是基于同步I/O机制的,所以libevent实际是一个基于同步I/O机制的库. 对于I/O设计模式,与Reactor相对应的还有Proactor.下边我们先来看下这两者的不同之处. Rea

高性能IO之Reactor模式

讲到高性能IO绕不开Reactor模式,它是大多数IO相关组件如Netty.Redis在使用的IO模式,为什么需要这种模式,它是如何设计来解决高性能并发的呢? 最最原始的网络编程思路就是服务器用一个while循环,不断监听端口是否有新的套接字连接,如果有,那么就调用一个处理函数处理,类似:while(true){ socket = accept(); handle(socket) } 这种方法的最大问题是无法并发,效率太低,如果当前的请求没有处理完,那么后面的请求只能被阻塞,服务器的吞吐量太低.

两种高效的事件处理模型:Reactor模式和Proactor模式

随着IO多路复用技术的出现,出现了很多事件处理模式.同步I/O模型通常由Reactor模式实现,而异步I/O模型则由Proactor模式实现. Reactor模式: Reator类图如上所示,Reactor模式又叫反应器或反应堆,即实现注册描述符(Handle)及事件的处理器(EventHandler),当有事件发生的时候,事件多路分发器(Event Demultiplexer)做出反应,调用事件具体的处理函数(ConcreteEventHandler::handle_event()). Rea

C++服务器设计(一):基于I/O复用的Reactor模式

I/O模型选择 在网络服务端编程中,一个常见的情景是服务器需要判断多个已连接套接字是否可读,如果某个套接字可读,则读取该套接字数据,并进行进一步处理. 在最常用的阻塞式I/O模型中,我们对每个连接套接字通过轮流read系统调用获取可读数据.如图3-1所示,read系统调用将会把该线程阻塞,直到数据报到达且被复制到应用进程的缓冲区中时才会返回. 图3-1 阻塞式I/O模型 在阻塞式I/O模型中,数据可读和读取数据这两个操作被合并在了一个系统调用中,对于单个套接字是否可读的判断,必须要等到实际数据接

Reactor模式

对象行为类的设计模式,对同步事件分拣和派发.别名Dispatcher(分发器) Reactor模式是处理并发I/O比较常见的一种模式,用于同步I/O,中心思想是将所有要处理的I/O事件注册到一个中心I/O多路复用器上,同时主线程阻塞在多路复用器上:一旦有I/O事件到来或是准备就绪(区别在于多路复用器是边沿触发还是水平触发),多路复用器返回并将相应I/O事件分发到对应的处理器中. Reactor的事件处理机制 普通函数调用的机制:程序调用某函数->函数执行,程序等待->函数将结果和控制权返回给程

(转)reactor模式

转自: http://www.blogjava.net/DLevin/archive/2015/09/02/427045.html Reactor模式详解 前记 第一次听到Reactor模式是三年前的某个晚上,一个室友突然跑过来问我什么是Reactor模式?我上网查了一下,很多人都是给出NIO中的 Selector的例子,而且就是NIO里Selector多路复用模型,只是给它起了一个比较fancy的名字而已,虽然它引入了EventLoop概 念,这对我来说是新的概念,但是代码实现却是一样的,因而

ACE - Reactor模式源码剖析及具体实现(大量源码慎入)

原文出自http://www.cnblogs.com/binchen-china,禁止转载. 在之前的文章中提到过Reactor模式和Preactor模式,现在利用ACE的Reactor来实现一个基于Reactor框架的服务器. 首先回顾下Reactor模式和Preactor模式. Reactor模式: Reactor模式实现非常简单,使用同步IO模型,即业务线程处理数据需要主动等待或询问,主要特点是利用epoll监听listen描述符是否有响应,及时将客户连接信息放于一个队列,epoll和队列