Reactor Pattern （二）
?
?
class DemultiPlexer extends Thread {private Selector selector;private Dispatcher dispatcher;@Overridepublic void run() {for (;;) {try {signal();} catch (IOException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}}private void signal() throws IOException, InterruptedException {int c = selector.select();if (c <= 0) {return;}Iterator<SelectionKey> keys = selector.selectedKeys().iterator();while (keys.hasNext()) {SelectionKey key = keys.next();keys.remove();if (key.isValid()) {if (key.isAcceptable()) {accept(key);} else if (key.isReadable()) {dispatcher.register(key);}}}}private void accept(SelectionKey key) throws IOException {ServerSocketChannel ssc = (ServerSocketChannel) key.channel();SocketChannel channel = ssc.accept();channel.configureBlocking(false);channel.register(selector, SelectionKey.OP_READ);//key.cancel();}public DemultiPlexer(Selector selector, Dispatcher dispatcher) {this.selector = selector;this.dispatcher = dispatcher;}}?
class Dispatcher extends Thread { private static LinkedBlockingQueue<SelectionKey> queue = new LinkedBlockingQueue<SelectionKey>();public void run() {for (;;) {try {dispatch();} catch (IOException e) {e.printStackTrace();} catch (InterruptedException e) { e.printStackTrace(); }}}public void dispatch() throws IOException, InterruptedException {synchronized (queue) {if (queue.size() <= 0) {queue.wait();} else {? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? //拿出对列中的第一个事件，创建一个新的任务处理程序? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? SelectionKey key = queue.poll(); new RequestHandler(key).start();}}}? ? ? /**? ? ? ?* 注册新的事件，并唤醒当前线程。? ? ? ?*/? ? ? ? public void register(SelectionKey key) throws IOException {synchronized (queue) {queue.add(key);queue.notify();} } }
?
?
请求处理程序class RequestHandler extends Thread {private SelectionKey selectionKey;private static Charset utf8 = Charset.forName("utf-8");@Overridepublic void run() {SocketChannel socketChannel = (SocketChannel) selectionKey.channel();ByteBuffer buff = ByteBuffer.allocate(128);try { // make sure the socketchannel is connected. if (!socketChannel.isConnected())socketChannel.finishConnect(); // read the data and print it after decode.while (socketChannel.read(buff) > 0) {buff.flip();System.out.println(utf8.decode(buff));buff.clear();}//remove the key for next selection operationselectionKey.cancel(); } catch (IOException e) {e.printStackTrace();}}public RequestHandler(SelectionKey selectionKey) {this.selectionKey = selectionKey;}}
?
?
请求客户端class Request extends Thread {private int port = 9000;@Overridepublic void run() {Charset utf8 = Charset.forName("utf-8");try {SocketChannel channel = SocketChannel.open(new InetSocketAddress(port));channel.configureBlocking(false); Selector selector = Selector.open();channel.register(selector, SelectionKey.OP_CONNECT);if (!channel.isConnected())channel.finishConnect();channel.write(utf8.encode("Hello non-blocking IO reactor pattern.."));channel.close();} catch (IOException e) {e.printStackTrace();}}public static void main(String args[]) {for (int i = 0; i < 50; i++)new Request().start();}}?
最后启动一个服务器端，来运行这个non-blocking IO的服务器端。在main函数中，首先open一个selector和ServerSocketChannel，配置为non-blocking IO模式，绑定服务端口9000并注册OP_ACCEPT。最后启动多路复用分发器线程和分配器线程。
?
public class ReactorPattern {public static void main(String args[]) throws IOException {Selector sel = Selector.open();ServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);ssc.socket().bind(new InetSocketAddress(9000));ssc.register(sel, SelectionKey.OP_ACCEPT);Dispatcher disp = new Dispatcher();disp.start();new DemultiPlexer(sel, disp).start();}}?小结在non-blocking IO模式中，DemultiPlexer类是担任了筛选事件和注册事件的工作，Selector.select()采用的是blocking方式，调用select方法，都必须等到其有返回值为至，但select方法本身是线程安全的，因此DemultiPlexer类在运行环境下可以存在多个线程，配合多个Dispacher线程一起工作。如
?
// initialize selector and register OP_ACCEPT......Dispatcher disp1 = new Dispatcher();disp1.start();new DemultiPlexer(sel, disp1).start();Dispatcher disp2 = new Dispatcher();disp2.start();new DemultiPlexer(sel, disp2).start();
?
预告Proactor模式是Reactor模式的兄弟，两者有很多相似之处，Proactor模式有何特点，相对于Reactor模式，它又有什么优点呢，如果您感兴趣，请继续关注我的博客。
?
?
[原创内容，版权所有，如有转载，请注明出处，如果您发现文中有什么错误请指出，不胜感激]
1 楼 Mojarra 2011-11-10 ps:Non-blocking模式中，Selector注册一个事件后，对这个事件做了适当的处理之后，必须在下一次select操作之前必须调用SelectionKey.cancel()方法。这一个特性限制了Selector在Reactor模式中的应用，试想一下，启动一个任务处理线程后，多路事件分发器在事件轮循中并不知道这个线程处理此事件是处于何种状态。这个事件到底有没有被处理完成，多路事件分发器是不清楚的，无论是SelectionKey，还是SocketChannel，都没有一个合适的方法知道它所代表的事件已经被处理。如果检测到一个key后，立即调用这个key的cancel方法，会发生一些意象不到的效果。
一个比较笨拙的办法是在Dispatcher中维护一个正在处理的事件的HashMap,在一个新的事件被注册到分配器时，首先检查这个事件的hash是否已存在于当前的hashmap中，如果存在则不启动新的任务处理线程，否则启动。事件处理完成后，从hashmap中删除此事件的hash。
还有一种方法是OP_ACCEPT事件从selector轮循中剥离出去，Selector只放在socketChannel中使用，ServerSocketChannel仍使用类似面向线程的socket编程方法，在ServerSocketChannel收到一个接入的连接时，启动一个线程，让socketChannel在线程中注册感兴趣的事件以及移除该事件。这样会导致一个任务处理线程开一个Selector实例，系统的开销增加。
这两种方法中，前者的开销小，编程复杂度低，整体上优于第二个方法。