摘要：基本上所有的網(wǎng)絡(luò)應(yīng)用都會(huì)示范一個(gè)的寫法。除了這些操作的主體是而不是，操作的是，而不是。以為例其過程是這樣的這段代碼就是創(chuàng)建一個(gè)，并注冊(cè)一個(gè)，并把附著到上。關(guān)鍵之一顯然是利用了協(xié)程的和，把回調(diào)轉(zhuǎn)換成順序的邏輯執(zhí)行。

基本上所有的網(wǎng)絡(luò)應(yīng)用都會(huì)示范一個(gè)tcp的echo寫法。前面我們已經(jīng)看到了如何使用協(xié)程和異步io來做tcp服務(wù)器的第一步，accept。下面是一個(gè)完整的echo server的實(shí)現(xiàn)（完整代碼）：

package org.github.taowen.daili;

import kilim.Pausable;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

public class Main {
    public static void main(String[] args) throws Exception {
        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.timeout = 5000;
                SocketChannel socketChannel = scheduler.accept(serverSocketChannel);
                socketChannel.configureBlocking(false);
                ByteBuffer byteBuffer = ByteBuffer.allocateDirect(1024);
                while (scheduler.read(socketChannel, byteBuffer) > 0) {
                    byteBuffer.flip();
                    scheduler.write(socketChannel, byteBuffer);
                    byteBuffer.clear();
                }
            }
        };
        scheduler.callSoon(task);
        scheduler.loop();
    }
}

從上面的代碼來看，完全沒有異步IO的感覺，代碼寫出來和傳統(tǒng)Java同步網(wǎng)絡(luò)編碼是一樣的。除了scheduler.accept,scheduler.read這些操作的主體是scheduler而不是socket，操作的是byte buffer，而不是input/output stream。
這段代碼中最關(guān)鍵的是其中的那個(gè)task，是一個(gè)協(xié)程。scheduler.accept，read和accept三處會(huì)引起task的跳出執(zhí)行，跳出的時(shí)候會(huì)把task當(dāng)前在做的IO等待記錄到內(nèi)部的一個(gè)叫SelectorBooking的身上。以readBlocked為例其過程是這樣的：

public int read(SocketChannel socketChannel, ByteBuffer byteBuffer) throws IOException, Pausable {
    int bytesCount = socketChannel.read(byteBuffer);
    if (bytesCount > 0) {
        return bytesCount;
    }
    SelectionKey selectionKey = socketChannel.keyFor(selector);
    if (null == selectionKey) {
        selectionKey = socketChannel.register(selector, SelectionKey.OP_READ);
        SelectorBooking booking = addSelectorBooking(selectionKey);
        selectionKey.attach(booking);
    } else {
        selectionKey.interestOps(selectionKey.interestOps() | SelectionKey.OP_READ);
    }
    SelectorBooking booking = (SelectorBooking) selectionKey.attachment();
    booking.readBlocked(getCurrentTimeMillis() + timeout);
    return socketChannel.read(byteBuffer);
}

這段代碼就是創(chuàng)建一個(gè)SelectorBooking，并注冊(cè)一個(gè)SelectionKey，并把booking附著到selection key上。這樣selection key被select出來之后，就可以根據(jù)booking找到對(duì)應(yīng)喚醒的task。注意的是selection key是一個(gè)socket一個(gè)的，但是可能對(duì)應(yīng)的有四個(gè)操作（accept/connect/read/write），所以booking上可能會(huì)有四個(gè)被阻塞掛起的task分別對(duì)應(yīng)不同的操作。
而booking和task的交互發(fā)生在booking.readBlocked這個(gè)調(diào)用內(nèi)部：

public void readBlocked(long deadline) throws Pausable {
    if (null != readTask) {
        throw new RuntimeException("multiple read blocked on same channel");
    }
    readDeadline = deadline;
    updateDeadline();
    readTask = Task.getCurrentTask();
    Task.pause(this);
    if (readDeadline == -1) {
        readUnblocked();
        throw new RuntimeException("timeout");
    }
}

其中 Task.getCurrentTask 是一個(gè)神奇的調(diào)用。它可以得到當(dāng)前的“協(xié)程”。得到的這個(gè)協(xié)程可以在掛起之后調(diào)用resume重新喚醒。
Task.pause 是另外一處神奇的調(diào)用。它使得當(dāng)前執(zhí)行的協(xié)程掛起。等到下面那行if被執(zhí)行到，已經(jīng)是別的地方調(diào)用resume之后的事情了。
通過這樣的一些列操作，就完成一個(gè)協(xié)程的掛起，并把協(xié)程和異步io等信息注冊(cè)到selector上的過程。
主循環(huán)只需要調(diào)用selector，找到就緒了的selection key，然后根據(jù)之前attach的附件找到booking，通過booking找到需要喚醒的協(xié)程，然后調(diào)用resume就可以讓協(xié)程上的業(yè)務(wù)邏輯繼續(xù)往下執(zhí)行了：

public void loop() {
    while (loopOnce()) {
    }
}

boolean loopOnce() {
    try {
        executeReadyTasks();
        doSelect();
        Iterator iterator = selector.selectedKeys().iterator();
        ioUnblocked(iterator);
        while (hasDeadSelectorBooking()) {
            SelectorBooking booking = selectorBookings.poll();
            booking.cancelDeadTasks(getCurrentTimeMillis());
        }
        return true;
    } catch (Exception e) {
        LOGGER.error("loop died", e);
        return false;
    }
}

這種做法非常經(jīng)典。關(guān)鍵之一顯然是利用了協(xié)程的pause和resume，把回調(diào)轉(zhuǎn)換成順序的邏輯執(zhí)行。關(guān)鍵之二就是利用了selection key的附件功能，把協(xié)程附著到了selection key上從而在select出來之后可以迅速恢復(fù)到阻塞之前的程序狀態(tài)（resume是一個(gè)局部上下文恢復(fù)的過程）。