单线程Reactor

package org.example.utils.echo.single;import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;public class EchoServerReactor implements Runnable{Selector selector;ServerSocketChannel serverSocketChannel;EchoServerReactor() throws IOException {//Reactor初始化selector = Selector.open();serverSocketChannel = ServerSocketChannel.open();InetSocketAddress address =new InetSocketAddress("localhost",8848);//非阻塞serverSocketChannel.configureBlocking(false);//分步处理,第一步,接收accept事件SelectionKey sk =serverSocketChannel.register(selector,0,new AcceptorHandler());// SelectionKey.OP_ACCEPTserverSocketChannel.socket().bind(address);System.out.println("服务端已经开始监听："+address);sk.interestOps(SelectionKey.OP_ACCEPT);}@Overridepublic void run() {try {while (!Thread.interrupted()){selector.select();Set<SelectionKey>  selected=selector.selectedKeys();Iterator<SelectionKey> it=selected.iterator();while (it.hasNext()){SelectionKey sk=it.next();dispatch(sk);}selected.clear();}} catch (IOException e) {throw new RuntimeException(e);}}private void dispatch(SelectionKey sk) {Runnable handler=(Runnable) sk.attachment();if (handler!=null){handler.run();}}class AcceptorHandler implements Runnable{@Overridepublic void run() {try {SocketChannel channel=serverSocketChannel.accept();if (channel!=null)new EchoHandler(selector,channel);} catch (IOException e) {throw new RuntimeException(e);}}}public static void main(String[] args) throws IOException {new Thread(new EchoServerReactor()).start();}
}

package org.example.utils.echo.single;import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;public class EchoHandler implements Runnable{final SocketChannel channel;final SelectionKey sk;final ByteBuffer byteBuffer=ByteBuffer.allocate(1024);static final int RECIEVING=0,SENDING=1;int state=RECIEVING;EchoHandler(Selector selector,SocketChannel c) throws IOException {channel=c;c.configureBlocking(false);sk=channel.register(selector,0);sk.attach(this);sk.interestOps(SelectionKey.OP_READ);selector.wakeup();}@Overridepublic void run() {try {if (state==SENDING){channel.write(byteBuffer);byteBuffer.clear();sk.interestOps(SelectionKey.OP_READ);state=RECIEVING;}else if (state==RECIEVING){int length=0;while ((length=channel.read(byteBuffer))>0){System.out.println(new String(byteBuffer.array(),0,length));}byteBuffer.flip();sk.interestOps(SelectionKey.OP_WRITE);state=SENDING;}} catch (IOException e) {throw new RuntimeException(e);}}
}

结果：

原理无非就是：

多线程，无非就是搞多个Reactor   ,   一个专门接受accept  ,  一个专门dispatch ,  再搞一个多线程池处理handle

这里面最主要的就是

handle类，sk.attach(this);把对象传回reactor

参考文献：

java高并发核心编程. 卷1，NIO、Netty、Redis、ZooKeeper  (尼恩)