单线程版本Reactor模式范例

Reactor反应器

/*** 单线程版的反应器模式demo*/
public class EchoServerReactor implements Runnable {Selector selector;ServerSocketChannel serverSocketChannel;public static void main(String[] args) {new Thread(new EchoServerReactor()).start();}void EchoServerReactor() throws Exception{//打开选择器selector = Selector.open();//ServerSocket连接监听通道serverSocketChannel = ServerSocketChannel.open();serverSocketChannel.socket().bind(new InetSocketAddress(NioDemoConfig.SOCKET_SERVER_IP, NioDemoConfig.SOCKET_SERVER_PORT));serverSocketChannel.configureBlocking(false);//注册serverSocket的accept事件SelectionKey sk = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);//将新连接处理器作为附件，绑定到sk选择键sk.attach(new AcceptorHandler());}@Overridepublic void run() {//选择器轮询try {while (!Thread.interrupted()) {selector.select();Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> it = selectionKeys.iterator();while (it.hasNext()) {//反应器负责dispatch收到的事件SelectionKey sk = it.next();dispatch(sk);}selectionKeys.clear();}} catch (IOException e) {e.printStackTrace();}}private void dispatch(SelectionKey sk) {Runnable handler = (Runnable) (sk.attachment());//调用之前绑定到选择键的handler处理器对象if (handler != null) {handler.run();}}//新连接处理器class AcceptorHandler implements Runnable{@Overridepublic void run() {//接受新连接try {SocketChannel channel = serverSocketChannel.accept();if (channel != null) {//需要为新连接创建一个输入输出的handler处理器new EchoHandler(selector, channel);}} catch (IOException e) {e.printStackTrace();}}}
}

Handler处理器

/*** 单线程版处理器*/
public class EchoHandler implements Runnable {final SocketChannel channel;final SelectionKey sk;final ByteBuffer buffer = ByteBuffer.allocate(NioDemoConfig.SEND_BUFFER_SIZE);static final int RECIEVING = 0, SENDING = 1;int state = RECIEVING;EchoHandler(Selector selector, SocketChannel c) throws IOException {channel = c;channel.configureBlocking(false);//取得选择键，设置感兴趣的IO事件sk = channel.register(selector, RECIEVING);//将处理器作为选择键的附件sk.attach(this);//注册Read就绪事件sk.interestOps(SelectionKey.OP_READ);selector.wakeup();}@Overridepublic void run() {try {if (state == RECIEVING) {//从通道读int length = 0;while ((length = channel.read(buffer)) > 0) {System.out.println(new String(buffer.array(), 0, length));}//读完后，准备开始写入通道，buffer切换为读模式buffer.flip();//读完后，注册write就绪事件sk.interestOps(SelectionKey.OP_WRITE);//读完后，进入发送状态state = SENDING;} else if (state == SENDING) {//写入通道channel.write(buffer);//写完后，准备开始从通道读，buffer切换为写模式buffer.clear();//写完后，注册read就绪事件sk.interestOps(SelectionKey.OP_READ);//写完后，进入接收状态state = RECIEVING;}//处理结束，不能关闭select_key，需要重复使用
//            sk.cancel();} catch (IOException e) {e.printStackTrace();}}
}