设计模式之Reactor(Dispatcher、Notifier)

Reactor设计模式

reactor设计模式，是一种基于事件驱动的设计模式。将一个或多个客户的服务请求分离（demultiplex）和调度（dispatch）给应用程序。

在Reactor模式中，有5个关键的参与者:

• 1.描述符（handle）：由操作系统提供，用于识别每一个事件，如Socket描述符、文件描述符等。在Linux中，它用一个整数来表示。事件可以来自外部，如来自客户端的连接请求、数据等。事件也可以来自内部，如定时器事件。
• 2.同步事件分离器（demultiplexer）：是一个函数，用来等待一个或多个事件的发生。调用者会被阻塞，直到分离器分离的描述符集上有事件发生。Linux的select函数是一个经常被使用的分离器。
• 3.事件处理器接口（event handler）：是由一个或多个模板函数组成的接口。这些模板函数描述了和应用程序相关的对某个事件的操作。
• 4.具体的事件处理器：是事件处理器接口的实现。它实现了应用程序提供的某个服务。每个具体的事件处理器总和一个描述符相关。它使用描述符来识别事件、识别应用程序提供的服务。
• 5.Reactor 管理器（reactor）：定义了一些接口，用于应用程序控制事件调度，以及应用程序注册、删除事件处理器和相关的描述符。它是事件处理器的调度核心。 Reactor管理器使用同步事件分离器来等待事件的发生。一旦事件发生，Reactor管理器先是分离每个事件，然后调度事件处理器，最后调用相关的模 板函数来处理这个事件

。

Reactor负责等待事件、分离事件和调度事件,实际上，Reactor管理器并没有被具体的 事件处理器调用，而是管理器调度具体的事件处理器，

图中的handle对应的是操作系统提供的句柄，例如I/O句柄，Event_Handler类持有这些句柄，

package com.jenny.model.callback.nioreactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
 * Created by jenny on 9/16/15.
 */
public class Reactor implements Runnable {

    private ServerSocketChannel serverSocketChannel = null;

    private Selector selector            = null;

    public Reactor() {
        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.socket().bind(new InetSocketAddress(8888));
            SelectionKey selectionKey = serverSocketChannel.register(selector,
                    SelectionKey.OP_ACCEPT);
            selectionKey.attach(new Acceptor());
            System.out.println("服务器启动正常!");
        } catch (IOException e) {
            System.out.println("启动服务器时出现异常!");
            e.printStackTrace();
        }
    }

    public void run() {
        while (true) {
            try {
                selector.select();

                Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                while (iter.hasNext()) {
                    SelectionKey selectionKey = iter.next();
                    dispatch((Runnable) selectionKey.attachment());
                    iter.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void dispatch(Runnable runnable) {
        if (runnable != null) {
            runnable.run();
        }
    }

    public static void main(String[] args) {
        new Thread(new Reactor()).start();
    }

    class Acceptor implements Runnable {
        public void run() {
            try {
                SocketChannel socketChannel = serverSocketChannel.accept();
                if (socketChannel != null) {
                    System.out.println("接收到来自客户端（"
                            + socketChannel.socket().getInetAddress().getHostAddress()
                            + "）的连接");
                    new Handler(selector, socketChannel);
                }

            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}

package com.jenny.model.callback.nioreactor;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;

/**
 * Created by jenny on 9/16/15.
 */
public class Handler implements Runnable {

    private static final int READ_STATUS  = 1;

    private static final int WRITE_STATUS = 2;

    private SocketChannel socketChannel;

    private SelectionKey selectionKey;

    private int status = READ_STATUS;

    public Handler(Selector selector, SocketChannel socketChannel) {
        this.socketChannel = socketChannel;
        try {
            socketChannel.configureBlocking(false);
            selectionKey = socketChannel.register(selector, 0);
            selectionKey.interestOps(SelectionKey.OP_READ);
            selectionKey.attach(this);
            selector.wakeup();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void run() {
        try {
            if (status == READ_STATUS) {
                read();
                selectionKey.interestOps(SelectionKey.OP_WRITE);
                status = WRITE_STATUS;
            } else if (status == WRITE_STATUS) {
                process();
                selectionKey.cancel();
                System.out.println("服务器发送消息成功!");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void read() throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        socketChannel.read(buffer);
        System.out.println("接收到来自客户端（" + socketChannel.socket().getInetAddress().getHostAddress()
                + "）的消息：" + new String(buffer.array()));
    }

    public void process() throws IOException {
        String content = "Hello World!";
        ByteBuffer buffer = ByteBuffer.wrap(content.getBytes());
        socketChannel.write(buffer);
    }
}

Hollywood principle:"Don't call me; I'll call you."