《Scalable IO in Java》笔记

Scalable IO in Java

http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf

基本上所有的网络处理程序都有以下基本的处理过程: 代码实现：

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();    }  }}

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(); }  }}

海外公司注册、海外银行开户、跨境平台代入驻、VAT、EPR等知识和在线办理：https://www.xlkjsw.com

原标题：《Scalable IO in Java》笔记

关键词：JAVA