摘要：接下來(lái)，就看怎么用協(xié)程來(lái)實(shí)現(xiàn)異步了。直接拿的原始寫代碼會(huì)死人的。引入?yún)f(xié)程就是為了把上下連續(xù)的業(yè)務(wù)邏輯放在一個(gè)協(xié)程里，把與業(yè)務(wù)關(guān)系不大的的處理部分放到框架的里。第三部分是放棄掉執(zhí)行權(quán)。這樣一個(gè)只能接收打印一行的異步應(yīng)用就寫好了。

前面已經(jīng)準(zhǔn)備好了greenlet對(duì)應(yīng)的Java版本了，一個(gè)刪減后的kilim（http://segmentfault.com/blog/taowen/1190000000697487）。接下來(lái)，就看怎么用協(xié)程來(lái)實(shí)現(xiàn)異步io了。首先，拿一段最最簡(jiǎn)單的tcp socket accept的代碼：

Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(9090));
serverSocketChannel.configureBlocking(false);
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("listening...");
selector.select();
scheduler.accept(serverSocketChannel);
System.out.println("hello");

這里使用的是java 6的NIO1的selector模型。直接拿NIO的原始api寫代碼會(huì)死人的。引入?yún)f(xié)程就是為了把上下連續(xù)的業(yè)務(wù)邏輯放在一個(gè)協(xié)程里，把與業(yè)務(wù)關(guān)系不大的selector的處理部分放到框架的ioloop里。也就是把一段交織的代碼，分成兩個(gè)關(guān)注點(diǎn)不同的組成部分。
改造之后的代碼在這里： https://github.com/taowen/daili/tree/1e319f929678213a8d8f63ee5e8b8cf016637317
這是改造之后的效果：

Scheduler scheduler = new Scheduler();
DailiTask task = new DailiTask(scheduler) {
    @Override
    public void execute() throws Pausable, Exception {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(9090));
        serverSocketChannel.configureBlocking(false);
        System.out.println("listening...");
        scheduler.accept(serverSocketChannel);
        System.out.println("hello");
    }
};
scheduler.callSoon(task);
scheduler.loop();

其中最關(guān)鍵的一行是 scheduler.accept(serverSocketChannel); 這個(gè)調(diào)用是阻塞的。但是只阻塞調(diào)用它的Task協(xié)程。如果有多個(gè)Task并行的話，別的Task可以在這個(gè)時(shí)候被運(yùn)行。那么scheduler.accept是如何做到把NIO的selector api轉(zhuǎn)換成這樣的形式的呢？

public SocketChannel accept(ServerSocketChannel serverSocketChannel) throws IOException, Pausable {
    SocketChannel socketChannel = serverSocketChannel.accept();
    if (null != socketChannel) {
        return socketChannel;
    }
    SelectionKey selectionKey = serverSocketChannel.keyFor(selector);
    if (null == selectionKey) {
        selectionKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, new WaitingSelectorIO());
    } else {
        selectionKey.interestOps(selectionKey.interestOps() | SelectionKey.OP_ACCEPT);
    }
    WaitingSelectorIO waitingSelectorIO = (WaitingSelectorIO) selectionKey.attachment();
    waitingSelectorIO.acceptBlockedAt = System.currentTimeMillis();
    waitingSelectorIO.acceptTask = (Task) Task.getCurrentTask();
    selectionKey.attach(waitingSelectorIO);
    Task.pause(waitingSelectorIO);
    return serverSocketChannel.accept();
}

這個(gè)函數(shù)分成四部分：第一部分是嘗試去accept，如果有戲就不用NIO了。第二部分是注冊(cè)selection key，說(shuō)明我希望知道什么時(shí)候可以accept了，并把task作為附件加上去。第三部分是Task.pause放棄掉執(zhí)行權(quán)。第四部分是task被回調(diào)了，說(shuō)明等待的accept已經(jīng)ok了，可以去調(diào)用了。
但是Task.pause了之后，是誰(shuí)在把這個(gè)暫停的task重新拉起來(lái)執(zhí)行的呢？這個(gè)就是scheduler的loop干的活了

public void loop() throws IOException {
    while (true) {
        executeReadyTasks();
        selector.select();
        Set selectionKeys = selector.selectedKeys();
        for (SelectionKey selectionKey : selectionKeys) {
            WaitingSelectorIO waitingSelectorIO = (WaitingSelectorIO) selectionKey.attachment();
            if (selectionKey.isAcceptable()) {
                Task taskToCall = waitingSelectorIO.acceptTask;
                waitingSelectorIO.acceptBlockedAt = 0;
                waitingSelectorIO.acceptTask = null;
                callSoon(taskToCall);
            }
        }
    }
}

在循環(huán)中調(diào)用selector.select獲得網(wǎng)絡(luò)事件的通知。如果selection key就緒了，就把附件里的task取出來(lái)回調(diào)。具體的回調(diào)發(fā)生在executeReadyTasks內(nèi)部，其實(shí)就是調(diào)用一下resume而已。

private void executeReadyTasks() {
    Task task;
    while((task = readyTasks.poll()) != null) {
        executeTask(task);
    }
}

private void executeTask(Task task) {
    try {
        task.resume();
    } catch (Exception e) {
        LOGGER.error("failed to execute task: " + task, e);
    }
}

這樣一個(gè)只能接收telnet 127.0.0.1 9090打印一行hello的異步io應(yīng)用就寫好了。