最近在学习NIO相关的知识部分，在学习的时候，发现Doug Lea大神的一个PPT地址，里面的内容写的非常好，
很值得学习，具体内容可以参见PPT和网上的文章，自己根据PPT中的例子和描述，自己将代码补充了一下。下面主要是PPT中的代码和实例。
1.Basic Reactor Design
package basic; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Set; /** * @ClassName: Reactor * @Description: 根据 Scalable IO in Java写的实例。 * @date 2017年3月7日 */ public class Reactor implements Runnable { final Selector selector; final ServerSocketChannel serverSocketChannel; public Reactor(int port) throws IOException { selector = Selector.open(); serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.socket().bind(new InetSocketAddress(port)); serverSocketChannel.configureBlocking(false); SelectionKey key = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); key.attach(new Acceptor()); } public void run() { try{ while(!Thread.interrupted()){ int n = selector.select();//等待事件 if(n == 0) continue; Set<SelectionKey> set = selector.selectedKeys(); Iterator<SelectionKey> iterator = set.iterator(); while(iterator.hasNext()){ dispatch(iterator.next()); } set.clear(); } }catch(Exception e){ e.printStackTrace(); } } private void dispatch(SelectionKey k) { Runnable runnable = (Runnable) k.attachment(); if(runnable != null) runnable.run(); } class Acceptor implements Runnable { public void run() { try { SocketChannel c = serverSocketChannel.accept();//接收请求 if(c != null){ System.out.println("New Connection ..."); new Handler(selector, c); } } catch (IOException e) { e.printStackTrace(); } } } final class Handler implements Runnable{ final SocketChannel socket; final SelectionKey key; ByteBuffer input = ByteBuffer.allocate(1024); ByteBuffer output = ByteBuffer.allocate(1024); static final int READING = 0,SENDING =1; int state = READING; public Handler(Selector selector, SocketChannel c) throws IOException { socket = c; socket.configureBlocking(false); key = socket.register(selector, 0); key.attach(this); key.interestOps(SelectionKey.OP_READ);//注册可读事件 selector.wakeup(); } public void run() { try { if(state == READING) read(); else if(state == SENDING) send(); } catch (Exception e) { e.printStackTrace(); } } boolean inputIsComplete(){ return input.hasRemaining(); } boolean outputIsComplete(){ return output.hasRemaining(); } void process(){ //读数据 StringBuilder reader = new StringBuilder(); input.flip(); while(input.hasRemaining()){ reader.append((char)input.get()); } System.out.println("[Client-INFO]"); System.out.println(reader.toString()); String str = "HTTP/1.1 200 OK\r\nDate: Fir, 10 March 2017 21:20:01 GMT\r\n"+ "Content-Type: text/html;charset=UTF-8\r\nContent-Length: 23\r\nConnection:close"+ "\r\n\r\nHelloWorld"+System.currentTimeMillis(); output.put(str.getBytes()); System.out.println("process over.... "); } void read() throws IOException{ socket.read(input); if(inputIsComplete()){ process(); state = SENDING; key.interestOps(SelectionKey.OP_WRITE); } } void send() throws IOException { output.flip(); socket.write(output); if(outputIsComplete()){ key.cancel(); } state = READING; key.interestOps(SelectionKey.OP_READ); } } public static void main(String[] args) throws IOException { new Thread(new Reactor(9001)).start(); System.out.println("Server start..."); } }
2.Per-State Handlers
package state; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Set; /** * @ClassName: Reactor * @Description: 根据 Scalable IO in Java写的实例。 * @author Lzyer * @date 2017年3月7日 */ public class Reactor implements Runnable { final Selector selector; final ServerSocketChannel serverSocketChannel; public Reactor(int port) throws IOException { selector = Selector.open(); serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.socket().bind(new InetSocketAddress(port)); serverSocketChannel.configureBlocking(false); SelectionKey key = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); key.attach(new Acceptor()); } public void run() { try{ while(!Thread.interrupted()){ int n = selector.select();//等待事件 if(n == 0) continue; Set<SelectionKey> set = selector.selectedKeys(); Iterator<SelectionKey> iterator = set.iterator(); while(iterator.hasNext()){ dispatch(iterator.next()); } set.clear(); } }catch(Exception e){ e.printStackTrace(); } } private void dispatch(SelectionKey k) { Runnable runnable = (Runnable) k.attachment(); if(runnable != null) runnable.run(); } class Acceptor implements Runnable { public void run() { try { SocketChannel c = serverSocketChannel.accept(); if(c != null){ System.out.println("New Connection ..."); new Handler2(selector, c); } } catch (IOException e) { e.printStackTrace(); } } } /*** * @ClassName: Handler2 * @Description: State-Object pattern * @author lzyer * @date 2017年3月8日 * */ class Handler2 implements Runnable{ final SocketChannel socket; final SelectionKey key; ByteBuffer input = ByteBuffer.allocate(1024); ByteBuffer output = ByteBuffer.allocate(1024); static final int READING = 0,SENDING =1; int state = READING; public Handler2(Selector selector, SocketChannel c) throws IOException { this.socket = c; this.socket.configureBlocking(false); key = socket.register(selector, SelectionKey.OP_READ); key.attach(this); selector.wakeup(); } public void run() { try { socket.read(input); if(inputIsComplete()){ process(); key.attach(new Sender());//send thread key.interestOps(SelectionKey.OP_WRITE); key.selector().wakeup(); } } catch (IOException e) { e.printStackTrace(); } } boolean inputIsComplete(){ return input.hasRemaining(); } boolean outputIsComplete(){ return output.hasRemaining(); } void process(){ //读数据 StringBuilder reader = new StringBuilder(); input.flip(); while(input.hasRemaining()){ reader.append((char)input.get()); } System.out.println("[Client-INFO]"); System.out.println(reader.toString()); String str = "HTTP/1.1 200 OK\r\nDate: Sat, 31 Dec 2005 23:59:59 GMT\r\n"+ "Content-Type: text/html;charset=UTF-8\r\nContent-Length: 23\r\nConnection:close"+ "\r\n\r\nWelcome NIO "+System.currentTimeMillis(); output.put(str.getBytes()); System.out.println("process over.... "); } class Sender implements Runnable{ public void run() { try { output.flip(); socket.write(output); if(outputIsComplete()) key.cancel(); socket.close();//记住要关闭 } catch (IOException e) { e.printStackTrace(); } } } } public static void main(String[] args) throws IOException { new Thread(new Reactor(9001)).start(); System.out.println("Server start..."); } }
3.Handler with Thread Pool
package pool; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /** * @ClassName: Reactor * @Description: 根据 Scalable IO in Java写的实例。 * @author Lzyer * @date 2017年3月7日 */ public class Reactor implements Runnable { final Selector selector; final ServerSocketChannel serverSocketChannel; public Reactor(int port) throws IOException { selector = Selector.open(); serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.socket().bind(new InetSocketAddress(port)); serverSocketChannel.configureBlocking(false); SelectionKey key = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); key.attach(new Acceptor()); } public void run() { try{ while(!Thread.interrupted()){ int n = selector.select(); if(n == 0) continue; Set<SelectionKey> set = selector.selectedKeys(); Iterator<SelectionKey> iterator = set.iterator(); while(iterator.hasNext()){ dispatch(iterator.next()); } set.clear(); } }catch(Exception e){ e.printStackTrace(); } } private void dispatch(SelectionKey k) { Runnable runnable = (Runnable) k.attachment(); if(runnable != null) runnable.run(); } class Acceptor implements Runnable { public void run() { try { SocketChannel c = serverSocketChannel.accept(); if(c != null){ System.out.println("New Connection ..."); new Handler3(selector, c); } } catch (IOException e) { e.printStackTrace(); } } } /** * @ClassName: Handler3 * @Description: 使用线程池 * @author Lzyer * @date 2017年3月8日 * */ class Handler3 implements Runnable { final SocketChannel socket; final SelectionKey key; ByteBuffer input = ByteBuffer.allocate(1024); ByteBuffer output = ByteBuffer.allocate(1024); static final int READING = 0,SENDING =1, PROCESSING =3; int state = READING; ExecutorService pool = Executors.newCachedThreadPool(); public Handler3(Selector selector, SocketChannel c) throws IOException { this.socket = c; this.socket.configureBlocking(false); key = socket.register(selector, SelectionKey.OP_READ); key.attach(this); selector.wakeup(); } public void run() { try { if(state == READING) read(); else if(state == SENDING) send(); } catch (IOException e) { e.printStackTrace(); } } synchronized void read() throws IOException { socket.read(input); if (inputIsComplete()) { state = PROCESSING; pool.execute(new Processer()); } } void send() throws IOException { output.flip(); socket.write(output); if(outputIsComplete()){ key.cancel(); } state = READING; key.interestOps(SelectionKey.OP_READ); } void process(){ //读数据 StringBuilder reader = new StringBuilder(); input.flip(); while(input.hasRemaining()){ reader.append((char)input.get()); } System.out.println("[Client-INFO]"); System.out.println(reader.toString()); String str = "HTTP/1.1 200 OK\r\nDate: Fir, 10 March 2017 21:20:01 GMT\r\n"+ "Content-Type: text/html;charset=UTF-8\r\nContent-Length: 32\r\nConnection:close"+ "\r\n\r\nWelcome JAVA World "+System.currentTimeMillis(); output.put(str.getBytes()); System.out.println("process .... "); } boolean inputIsComplete(){ return input.hasRemaining(); } boolean outputIsComplete(){ return output.hasRemaining(); } synchronized void processAndHandOff() { process(); state = SENDING; // or rebind attachment key.interestOps(SelectionKey.OP_WRITE); System.out.println("processer over...."); } class Processer implements Runnable { public void run() { processAndHandOff(); } } } public static void main(String[] args) throws IOException { new Thread(new Reactor(9001)).start(); System.out.println("Server start..."); } }
4Using Multiple Reactors
package multiple; import java.io.IOException; import java.util.concurrent.atomic.AtomicInteger; public class Server { private static Reactor mainReactor;//主要负责接收请求 private static Reactor[] subReactors;//负责处理IO private static final int SUB_SIZE = 3; private static final int port = 1234; private static AtomicInteger nextIndex = new AtomicInteger(); public static Reactor nextSubReactor(){ long nextIndexValue = nextIndex.getAndIncrement(); if(nextIndexValue < 0){ nextIndex.set(0); nextIndexValue = 0; } return subReactors[(int) (nextIndexValue%subReactors.length)]; } public static void main(String[] args) { try { mainReactor = new Reactor(port, true); subReactors = new Reactor[SUB_SIZE]; for(int i=0; i< subReactors.length; i++){ subReactors[i] = new Reactor(port, false); } mainReactor.start(); for(int i=0; i< subReactors.length; i++){ subReactors[i].start(); } } catch (IOException e) { e.printStackTrace(); } } }
package multiple; 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.util.Iterator; /** * @ClassName: Reactor * @Description: 根据 Scalable IO in Java写的实例。 * @author Lzyer * @date 2017年3月10日 */ public class Reactor extends Thread { final Selector selector; final boolean isMain;//主从的标志 public Reactor(int port, boolean isMain) throws IOException { this.isMain = isMain; selector = Selector.open(); System.out.println(selector +" isMainReactor = "+isMain); if(isMain){ ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.socket().bind(new InetSocketAddress(port)); serverSocketChannel.configureBlocking(false); SelectionKey key = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); key.attach(new Acceptor(selector, serverSocketChannel)); selector.wakeup(); System.out.println(getClass().getSimpleName()+" start on "+port+" ...\n"); } } public void run() { try { while(!Thread.interrupted()){ int n = selector.select(10);//会阻塞导致不能register，设置阻塞时间 if(n == 0) continue; Iterator<SelectionKey> iterator = selector.selectedKeys().iterator(); while(iterator.hasNext()){ dispatch(iterator.next()); iterator.remove(); } } } catch (IOException e) { e.printStackTrace(); } } private void dispatch(SelectionKey k) { Runnable runnable = (Runnable) k.attachment(); if(runnable != null) runnable.run(); } }
package multiple; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; /** * @ClassName: Acceptor * @Description: 接收请求 * @author Lzyer * @date 2017年3月9日 * */ public class Acceptor implements Runnable { private ServerSocketChannel serverSocketChannel; private Selector selector; public Acceptor(Selector selector,ServerSocketChannel ser) { this.selector = selector; this.serverSocketChannel = ser; } public void run() { try{ System.out.println(selector+" accept ...."); SocketChannel socketChannel = serverSocketChannel.accept(); if(socketChannel != null){ /** * 开启了多个reactor池,一个selector负责接收和处理IO * new Handler(selector, socketChannel); */ //使用的主从结构,一个主selector负责接收,其他的负责处理IO Selector selector2 = Server.nextSubReactor().selector; System.out.println(selector2+" clientChannel not null..."+socketChannel); new Handler(selector2, socketChannel); } }catch(Exception e){ e.printStackTrace(); } } }
package multiple; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; /** * @ClassName: Handler * @Description: 处理IO操作 * @author Lzyer * @date 2017年3月9日 * */ public class Handler implements Runnable { final SocketChannel socket; final SelectionKey key; ByteBuffer input = ByteBuffer.allocate(1024); ByteBuffer output = ByteBuffer.allocate(1024); static final int READING = 0,SENDING =1; int state = READING; public Handler(Selector selector, SocketChannel c) throws IOException { socket = c; socket.configureBlocking(false); key = socket.register(selector,SelectionKey.OP_READ); key.attach(this); selector.wakeup(); System.out.println(selector+" connect success..."); } public void run() { try { if(state == READING) read(); else if(state == SENDING) send(); } catch (Exception e) { e.printStackTrace(); } } boolean inputIsComplete(){ return input.hasRemaining(); } boolean outputIsComplete(){ return output.hasRemaining(); } void process(){ //读数据 StringBuilder reader = new StringBuilder(); input.flip(); while(input.hasRemaining()){ reader.append((char)input.get()); } System.out.println("[Client-INFO]"); System.out.println(reader.toString()); String str = "HTTP/1.1 200 OK\r\nDate: Fir, 10 March 2017 21:20:01 GMT\r\n"+ "Content-Type: text/html;charset=UTF-8\r\nContent-Length: 24\r\nConnection:close"+ "\r\n\r\nHelloRector"+System.currentTimeMillis(); output.put(str.getBytes()); System.out.println("process .... "); } void read() throws IOException{ socket.read(input); if(inputIsComplete()){ process(); state = SENDING; key.interestOps(SelectionKey.OP_WRITE); } } void send() throws IOException { output.flip(); socket.write(output); if(outputIsComplete()){ key.cancel(); } state = READING; key.interestOps(SelectionKey.OP_READ); } }
代码地址：链接
参考:http://blog.csdn.net/jjzhk/article/details/39553613
http://www.iclojure.com/blog/articles/2015/12/05/Scalable-IO-in-Java
