Scalable IO in Java
大部分IO都是下面这个步骤,
Most have same basic structure:
Read request Decode request Process service Encode reply Send reply关键是如何处理并发, 最原始就是单纯的用多线程
class Server implements Runnable { public void run() { try { ServerSocket ss = new ServerSocket(PORT); while (!Thread.interrupted()) new Thread(new Handler(ss.accept())).start(); //创建新线程来handle // or, single-threaded, or a thread pool } catch (IOException ex) { /* ... */ } } static class Handler implements Runnable { final Socket socket; Handler(Socket s) { socket = s; } public void run() { try { byte[] input = new byte[MAX_INPUT]; socket.getInputStream().read(input); byte[] output = process(input); socket.getOutputStream().write(output); } catch (IOException ex) { /* ... */ } } private byte[] process(byte[] cmd) { /* ... */ } }} 显然简单的多线程会带来扩展性问题, 当client数量变的很多的时候, 还其他的可用性, 性能的问题
解决方法就是Divide-and-conquer, 分开后, 就需要Event-driven Designs来串联起来...
单线程版本的Reactor, 所有事情read, process, write都由单个线程完成, 完成一步重新设置下一步的event, 然后干其他的事
问题当然就是, 其中任何步骤不能消耗太多时间, 因为只有一个线程, 你占住了就会block其他任务 ps, 不明白为什么要画那么大个acceptor, 只是作为第一步的callback对象...
看代码会更清楚,
class Reactor implements Runnable { final Selector selector; final ServerSocketChannel serverSocket; Reactor(int port) throws IOException { //Reactor初始化 selector = Selector.open(); serverSocket = ServerSocketChannel.open(); serverSocket.socket().bind(new InetSocketAddress(port)); serverSocket.configureBlocking(false); //非阻塞 SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT); //分步处理,第一步,接收accept事件 sk.attach(new Acceptor()); //attach callback object, Acceptor } public void run() { try { while (!Thread.interrupted()) { selector.select(); Set selected = selector.selectedKeys(); Iterator it = selected.iterator(); while (it.hasNext()) dispatch((SelectionKey)(it.next()); //Reactor负责dispatch收到的事件 selected.clear(); } } catch (IOException ex) { /* ... */ } } void dispatch(SelectionKey k) { Runnable r = (Runnable)(k.attachment()); //调用之前注册的callback对象 if (r != null) r.run(); } class Acceptor implements Runnable { // inner public void run() { try { SocketChannel c = serverSocket.accept(); if (c != null) new Handler(selector, c); } catch(IOException ex) { /* ... */ } } }}final class Handler implements Runnable { final SocketChannel socket; final SelectionKey sk; ByteBuffer input = ByteBuffer.allocate(MAXIN); ByteBuffer output = ByteBuffer.allocate(MAXOUT); static final int READING = 0, SENDING = 1; int state = READING; Handler(Selector sel, SocketChannel c) throws IOException { socket = c; c.configureBlocking(false); // Optionally try first read now sk = socket.register(sel, 0); sk.attach(this); //将Handler作为callback对象 sk.interestOps(SelectionKey.OP_READ); //第二步,接收Read事件 sel.wakeup(); } boolean inputIsComplete() { /* ... */ } boolean outputIsComplete() { /* ... */ } void process() { /* ... */ } public void run() { try { if (state == READING) read(); else if (state == SENDING) send(); } catch (IOException ex) { /* ... */ } } void read() throws IOException { socket.read(input); if (inputIsComplete()) { process(); state = SENDING; // Normally also do first write now sk.interestOps(SelectionKey.OP_WRITE); //第三步,接收write事件 } } void send() throws IOException { socket.write(output); if (outputIsComplete()) sk.cancel(); //write完就结束了, 关闭select key }}//上面 的实现用Handler来同时处理Read和Write事件, 所以里面出现状态判断//我们可以用State-Object pattern来更优雅的实现class Handler { // ... public void run() { // initial state is reader socket.read(input); if (inputIsComplete()) { process(); sk.attach(new Sender()); //状态迁移, Read后变成write, 用Sender作为新的callback对象 sk.interest(SelectionKey.OP_WRITE); sk.selector().wakeup(); } } class Sender implements Runnable { public void run(){ // ... socket.write(output); if (outputIsComplete()) sk.cancel(); } }}
单线程模式的局限还是比较明显的
所以改进是, 将比较耗时的部分, 从reactor线程中分离出去, 让reactor专门负责IO 而另外创建Thread Pool和queue来缓存和处理任务 所以其实已经进化成Proactor模式, 异步模式
class Handler implements Runnable { // uses util.concurrent thread pool static PooledExecutor pool = new PooledExecutor(...); static final int PROCESSING = 3; // ... synchronized void read() { // ... socket.read(input); if (inputIsComplete()) { state = PROCESSING; pool.execute(new Processer()); //使用线程pool异步执行 } } synchronized void processAndHandOff() { process(); state = SENDING; // or rebind attachment sk.interest(SelectionKey.OP_WRITE); //process完,开始等待write事件 } class Processer implements Runnable { public void run() { processAndHandOff(); } }} 使用多个reactor进程, 主reactor只负责accept, 然后将接收到的socketchannel交给subReactor去listen和处理
当然也可以在subReactor下加上线程池进行异步处理 坦白的说, 没看出用多个reactor有啥大的提升, 降低mainReactor listen的负担?
Selector[] selectors; //subReactors集合, 一个selector代表一个subReactorint next = 0;class Acceptor { // ... public synchronized void run() { ... Socket connection = serverSocket.accept(); //主selector负责accept if (connection != null) new Handler(selectors[next], connection); //选个subReactor去负责接收到的connection if (++next == selectors.length) next = 0; }}