Java NIO 非阻塞Socket服务器构建

推荐阅读IBM developerWorks中NIO的入门教程，尤其是对块I/O和流I/O不太清楚的开发者。

说到socket服务器，第一反应是java.net.Socket这个类。事实上在并发和响应时间要求不高的场合，是可以用java.net.Socket来实现的，比如写一个局域网聊天工具、发送文件等。但它的缺点也很明显，需要自行对接受的线程进行维护，管理缓冲区的分配等，我尝试过用java.net.Socket完成一个瞬时负载在千人左右的服务器，却因后期改动和维护异常麻烦而放弃。

Java自1.4以后，加入了新IO特性，这便是本文要介绍的NIO。下面是一段服务器的示例代码（引用自xpbug的Blog）：

public class EchoServer {
    public static SelectorLoop connectionBell;
    public static SelectorLoop readBell;
    public boolean isReadBellRunning=false;
 
    public static void main(String[] args) throws IOException {
        new EchoServer().startServer();
    }
     
    // 启动服务器
    public void startServer() throws IOException {
        // 准备好一个闹钟.当有链接进来的时候响.
        connectionBell = new SelectorLoop();
         
        // 准备好一个闹装,当有read事件进来的时候响.
        readBell = new SelectorLoop();
         
        // 开启一个server channel来监听
        ServerSocketChannel ssc = ServerSocketChannel.open();
        // 开启非阻塞模式
        ssc.configureBlocking(false);
         
        ServerSocket socket = ssc.socket();
        socket.bind(new InetSocketAddress("localhost",7878));
         
        // 给闹钟规定好要监听报告的事件,这个闹钟只监听新连接事件.
        ssc.register(connectionBell.getSelector(), SelectionKey.OP_ACCEPT);
        new Thread(connectionBell).start();
    }
     
    // Selector轮询线程类
    public class SelectorLoop implements Runnable {
        private Selector selector;
        private ByteBuffer temp = ByteBuffer.allocate(1024);
         
        public SelectorLoop() throws IOException {
            this.selector = Selector.open();
        }
         
        public Selector getSelector() {
            return this.selector;
        }
 
        @Override
        public void run() {
            while(true) {
                try {
                    // 阻塞,只有当至少一个注册的事件发生的时候才会继续.
                    this.selector.select();
                     
                    Set<SelectionKey> selectKeys = this.selector.selectedKeys();
                    Iterator<SelectionKey> it = selectKeys.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        it.remove();
                        // 处理事件. 可以用多线程来处理.
                        this.dispatch(key);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
         
        public void dispatch(SelectionKey key) throws IOException, InterruptedException {
            if (key.isAcceptable()) {
                // 这是一个connection accept事件, 并且这个事件是注册在serversocketchannel上的.
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                // 接受一个连接.
                SocketChannel sc = ssc.accept();
                 
                // 对新的连接的channel注册read事件. 使用readBell闹钟.
                sc.configureBlocking(false);
                sc.register(readBell.getSelector(), SelectionKey.OP_READ);
                 
                // 如果读取线程还没有启动,那就启动一个读取线程.
                synchronized(EchoServer.this) {
                    if (!EchoServer.this.isReadBellRunning) {
                        EchoServer.this.isReadBellRunning = true;
                        new Thread(readBell).start();
                    }
                }
                 
            } else if (key.isReadable()) {
                // 这是一个read事件,并且这个事件是注册在socketchannel上的.
                SocketChannel sc = (SocketChannel) key.channel();
                // 写数据到buffer
                int count = sc.read(temp);
                if (count < 0) {
                    // 客户端已经断开连接.
                    key.cancel();
                    sc.close();
                    return;
                }
                // 切换buffer到读状态,内部指针归位.
                temp.flip();
                String msg = Charset.forName("UTF-8").decode(temp).toString();
                System.out.println("Server received ["+msg+"] from client address:" + sc.getRemoteAddress());
                 
                Thread.sleep(1000);
                // echo back.
                sc.write(ByteBuffer.wrap(msg.getBytes(Charset.forName("UTF-8"))));
                 
                // 清空buffer
                temp.clear();
            }
        }
    }
 
}

 此外，还有一个看上去更“规范”的示例《ServerSocketChannel与SocketChannel的使用》，这里就不再引用了。

OK，原文的注释已经很详细了，这里进一步解析这段代码。

首先是java.nio.channels.ServerSocketChannel这个类，引用官方的描述：

public abstract class ServerSocketChannelextends AbstractSelectableChannel

A selectable channel for stream-oriented listening sockets.

Server-socket channels are not a complete abstraction of listening network sockets. Binding and the manipulation of socket options must be done through an associated ServerSocket object obtained by invoking the socket method. It is not possible to create a channel for an arbitrary, pre-existing server socket, nor is it possible to specify the SocketImpl object to be used by a server socket associated with a server-socket channel.

A server-socket channel is created by invoking the open method of this class. A newly-created server-socket channel is open but not yet bound. An attempt to invoke the accept method of an unbound server-socket channel will cause a NotYetBoundException to be thrown. A server-socket channel can be bound by invoking one of the bind methods of an associated server socket.

Server-socket channels are safe for use by multiple concurrent threads.

再看选择器Selector的用法：请戳这里。