【精尽Netty源码解析】1.Scalable IO in Java——多Reactor的代码实现

public class Server implements Runnable{ private final int port;
private ServerSocket serverSocket; public Server(int port){
this.port = port;
try {
this.serverSocket = new ServerSocket(port);
} catch (IOException e) {
e.printStackTrace();
}
} @Override
public void run(){
try {
while (!Thread.interrupted()) {
new Thread(new Handler(serverSocket.accept())).start();
}
} catch (IOException e) {
e.printStackTrace();
}
} public int getPort() {
return port;
} public static void main(String[] args) {
new Thread(new Server(9001)).start();
}}


public class Handler implements Runnable{ private final Socket clientSocket; public Handler(Socket clientSocket){
this.clientSocket = clientSocket;
} @Override
public void run() {
int readSize;
byte[] readBuf = new byte[BUF_SIZE];
try {
InputStream in = clientSocket.getInputStream();
OutputStream out = clientSocket.getOutputStream();
while ((readSize = in.read(readBuf)) != -1) {
out.write(readBuf, 0, readSize);
}
} catch (Exception e) {
e.printStackTrace();
}
}}


public abstract class Reactor extends Thread{ protected final int port;
protected final ServerSocketChannel serverChannel;
protected final boolean isMainReactor;
protected final boolean useMultipleReactors;
protected final long timeout;
protected Selector selector; public Reactor(int port, ServerSocketChannel serverChannel, boolean isMainReactor, boolean useMultipleReactors, long timeout){
this.port = port;
this.serverChannel = serverChannel;
this.isMainReactor = isMainReactor;
this.useMultipleReactors = useMultipleReactors;
this.timeout = timeout;
} @Override
public void run(){
try {
init();
while(!Thread.interrupted()){
//不可以使用阻塞的select方式，否则accept后subReactor的selector在register的时候会一直阻塞
//但是修改为带有超时的select或者selectNow后，subReactor的selector在register就不会阻塞了
//最终选择了带有超时的select是因为使用selectNow的无限循环会导致CPU飙高特别快
//并且如果使用阻塞的select方式，还需要知道在哪里调用wakeup，否则会一直阻塞，使用非阻塞方式就不需要wakeup了
//selector.select();
//if(selector.selectNow() > 0){
if(selector.select(timeout) > 0){
log(selector+" isMainReactor="+isMainReactor+" select...");
Iterator<SelectionKey> keyIt = selector.selectedKeys().iterator();
while(keyIt.hasNext()){
SelectionKey key = keyIt.next();
dispatch(key);
keyIt.remove();
}
}
}
log(getClass().getSimpleName()+" end on "+port+" ..."+"\n");
} catch (IOException e) {
e.printStackTrace();
}
} private void init() throws IOException{
selector = Selector.open();
log(selector+" isMainReactor="+isMainReactor); if(isMainReactor){
//serverChannel = ServerSocketChannel.open();
serverChannel.socket().bind(new InetSocketAddress(port));
serverChannel.configureBlocking(false);
SelectionKey key = serverChannel.register(selector, SelectionKey.OP_ACCEPT);
key.attach(newAcceptor(selector));
log(getClass().getSimpleName()+" start on "+port+" ..."+"\n");
}else{ } //如果使用阻塞的select方式，且开启下面的代码的话，相当于开启了多个reactor池，而不是mainReactor和subReactor的关系了
//SelectionKey key = serverChannel.register(selector, SelectionKey.OP_ACCEPT);
//key.attach(newAcceptor(selector, serverChannel));
} public abstract Acceptor newAcceptor(Selector selector); /**
* 事件和事件处理器的绑定
* <ul>
* <li>管理IO读/写事件到事件处理器的一一对应的绑定</li>
* </ul>
*/
private void dispatch(SelectionKey key){
Runnable r = (Runnable)key.attachment();
if(r != null){
r.run();
}
}}


public abstract class Acceptor extends Thread { protected final Selector selector;
protected final ServerSocketChannel serverChannel;
protected final boolean useMultipleReactors; public Acceptor(Selector selector, ServerSocketChannel serverChannel, boolean useMultipleReactors){
this.selector = selector;
this.serverChannel = serverChannel;
this.useMultipleReactors = useMultipleReactors;
} @Override
public void run() {
log(selector+" accept...");
try {
SocketChannel clientChannel = serverChannel.accept();
if(clientChannel != null){
log(selector+" clientChannel not null...");
//如果使用阻塞的select方式，且目的是开启了多个reactor池，而不是mainReactor和subReactor的关系的话，
//则下面就不是nextSubSelector().selector，而是改为传递当前实例的selector对象即可
handle(useMultipleReactors ? nextSubReactor().selector : selector, clientChannel);
}else{
log(selector+" clientChannel is null...");
}
} catch (IOException e) {
e.printStackTrace();
}
} /**
* 在每个具体的Handler下调用run方法是为了令其从connecting状态变为reading状态，
* 和原pdf版本下的做法是一样的，只不过原pdf版本是在构造函数直接修改设置了感兴趣为read事件
*/
public abstract void handle(Selector selector, SocketChannel clientSocket);}


public abstract class Handler extends Thread { private enum State{
CONNECTING(0),
READING(SelectionKey.OP_READ),
PROCESSING(2),
WRITING(SelectionKey.OP_WRITE); private final int opBit;
private State(int operateBit){
opBit = operateBit;
}
} private State state;
protected final SocketChannel clientChannel;
protected final SelectionKey key; protected final ByteBuffer readBuf;
protected final StringBuilder readData = new StringBuilder();
protected ByteBuffer writeBuf; public Handler(Selector selector, SocketChannel clientChannel){
this.state = State.CONNECTING;
SelectionKey key = null;
try {
clientChannel.configureBlocking(false);
//这里在使用subSelector的时候会阻塞，为什么？是因为使用了阻塞的select方法，非阻塞的才可以
//但如果使用reactor池的话，那是因为需要serverChannel注册selector的accept事件！？必须对应上才可以通过，否则阻塞
key = clientChannel.register(selector, this.state.opBit);
key.attach(this);
} catch (Exception e) {
e.printStackTrace();
}
this.clientChannel = clientChannel;
this.key = key;
this.readBuf = ByteBuffer.allocate(byteBufferSize());
log(selector+" connect success...");
} @Override
public void run() {
switch (state) {
case CONNECTING:
connect();
break;
case READING:
readAndProcess();
break;
case WRITING:
write();
break;
default:
err("\nUnsupported State: "+state+" ! overlap processing with IO...");
}
} private void connect() {
interestOps(State.READING);
} /**
* But harder to overlap processing with IO<br/>
* Best when can first read all input a buffer<br/>
* <br>
* That why we used synchronized on read method!<br/>
* Just to protected read buffer And handler state...<br/>
* <br>
* 其实就是害怕重叠IO和工作线程处理不一致：例如Reactor单线程读某个key的IO完毕后立马开启工作线程的处理，
* 紧接着Reactor单线程处理第二个IO key的时候发现还是之前的那个key的读IO事件，但是之前同一个key的处理还未完成，
* 不等待之前的处理完成的话，就会出现多个线程同时访问修改Handler里面数据的情况，导致出错，
* 但是最好先把数据都全部读入buffer中就可以规避了！？
*
* <p>此处的synchronized同步是为了防止state状态以及读写buffer在多线程访问中出现读脏数据，
* Debug调试的时候同时访问一个SelectionKey有2个线程：
* <br>1、Reactor单线程
* <br>2、读数据完毕后多线程处理的话，线程池里面执行processAndHandOff的线程
* <br>
* 不能单一使用volatile或者原子变量的原因是因为该方法为复合操作（check and act）
*/
private synchronized void readAndProcess(){
doRead();
doProcess();
} private void doRead(){
int readSize;
try {
while((readSize = clientChannel.read(readBuf)) > 0){
readData.append(new String(Arrays.copyOfRange(readBuf.array(), 0, readSize)));
readBuf.clear();
}
if(readSize == -1){
disconnect();
return;
}
} catch (IOException e) {
e.printStackTrace();
disconnect();
} log("readed from client:"+readData+", "+readData.length());
} private void doProcess(){
if(readIsComplete()){
state = State.PROCESSING;
processAndInterestWrite();
}
} /**
* 处理过程可能是比较耗时的，所以可考虑将其交由线程池处理，处理完毕后才注册感兴趣的write事件<p>
* 然而正是由于交由线程池处理所以可能造成重叠IO的多线程处理的状态问题，最好能一次性全部读入buffer，否则考虑同步状态处理问题
*/
private void processAndInterestWrite(){
Processor processor = new Processor();
if(useThreadPool){
execute(processor);
}else{
processor.run();
}
} private final class Processor implements Runnable{
@Override
public void run() {
processAndHandOff();
}
} private synchronized void processAndHandOff(){
if(process()){
interestOps(State.WRITING);
}
} //TODO 修改为复用output，即当output容量不足的时候就反复write，而不是每次都使用wrap来new一个新的
public boolean process(){
log("process readData="+readData.toString());
if(isQuit()){
disconnect();
return false;
} writeBuf = ByteBuffer.wrap(readData.toString().getBytes());
readData.delete(0, readData.length());
return true;
} private void write(){
try {
do{
clientChannel.write(writeBuf);
}while(!writeIsComplete());
} catch (IOException e) {
e.printStackTrace();
disconnect();
} String writeData = new String(Arrays.copyOf(writeBuf.array(), writeBuf.array().length));
log("writed to client:"+writeData+", "+writeData.length());
interestOps(State.READING);
} /**
* 事件和事件处理器的绑定
* <ul>
* <li>类似AWT中的addActionListener添加监听器/观察者</li>
* </ul>
* 不需要重置key的附件（key.attach）是因为key一直绑定使用的是当前this实例，
* 在Reactor dispatch的时候如果是接受（accept）该附件就是Acceptor实例，
* 否则就是绑定到该key的同一个Handler实例
*/
private void interestOps(State state){
this.state = state;
key.interestOps(state.opBit);
} public boolean isQuit(){
return false;
} private void disconnect(){
try {
clientChannel.close();
} catch (IOException e) {
e.printStackTrace();
}
log("\nclient Address=【"+clientAddress(clientChannel)+"】 had already closed!!! ");
} private static SocketAddress clientAddress(SocketChannel clientChannel){
return clientChannel.socket().getRemoteSocketAddress();
} public abstract int byteBufferSize(); public abstract boolean readIsComplete(); public abstract boolean writeIsComplete();}