使用Java NIO编写高性能的服务器

从JDK 1.4开始,Java的标准库中就包含了NIO,即所谓的“New IO”。其中最重要的功能就
首页 新闻资讯 行业资讯 使用Java NIO编写高性能的服务器

从JDK 1.4开始,Java的标准库中就包含了NIO,即所谓的“New IO”。其中最重要的功能就是提供了“非阻塞”的IO,当然包括了Socket。NonBlocking的IO就是对select(Unix平台下)以及 WaitForMultipleObjects(Windows平台)的封装,提供了高性能、易伸缩的服务架构。

说来惭愧,直到JDK1.4才有这种功能,但迟到者不一定没有螃蟹吃,NIO就提供了优秀的面向对象的解决方案,可以很方便地编写高性能的服务器。

话说回来,传统的Server/Client实现是基于Thread per request,即服务器为每个客户端请求建立一个线程处理,单独负责处理一个客户的请求。比如像Tomcat(新版本也会提供NIO方案)、Resin等Web服务器就是这样实现的。当然为了减少瞬间峰值问题,服务器一般都使用线程池,规定了同时并发的最大数量,避免了线程的无限增长。

但这样有一个问题:如果线程池的大小为100,当有100个用户同时通过HTTP现在一个大文件时,服务器的线程池会用完,因为所有的线程都在传输大文件了,即使第101个请求者仅仅请求一个只有10字节的页面,服务器也无法响应了,只有等到线程池中有空闲的线程出现。

另外,线程的开销也是很大的,特别是达到了一个临界值后,性能会显著下降,这也限制了传统的Socket方案无法应对并发量大的场合,而“非阻塞”的IO就能轻松解决这个问题。

下面只是一个简单的例子:服务器提供了下载大型文件的功能,客户端连接上服务器的12345端口后,就可以读取服务器发送的文件内容信息了。注意这里的服务器只有一个主线程,没有其他任何派生线程,让我们看看NIO是如何用一个线程处理N个请求的。

NIO服务器最核心的一点就是反应器模式:当有感兴趣的事件发生的,就通知对应的事件处理器去处理这个事件,如果没有,则不处理。所以使用一个线程做轮询就可以了。当然这里这是个例子,如果要获得更高性能,可以使用少量的线程,一个负责接收请求,其他的负责处理请求,特别是对于多CPU时效率会更高。

关于使用NIO过程中出现的问题,最为普遍的就是为什么没有请求时CPU的占用率为100%?出现这种问题的主要原因是注册了不感兴趣的事件,比如如果没有数据要发到客户端,而又注册了写事件(OP_WRITE),则在 Selector.select()上就会始终有事件出现,CPU就一直处理了,而此时select()应该是阻塞的。

另外一个值得注意的问题是:由于只使用了一个线程(多个线程也如此)处理用户请求,所以要避免线程被阻塞,解决方法是事件的处理者必须要即刻返回,不能陷入循环中,否则会影响其他用户的请求速度。

具体到本例子中,由于文件比较大,如果一次性发送整个文件(这里的一次性不是指send整个文件内容,而是通过while循环不间断的发送分组包),则主线程就会阻塞,其他用户就不能响应了。这里的解决方法是当有WRITE事件时,仅仅是发送一个块(比如4K字节)。发完后,继续等待WRITE事件出现,依次处理,直到整个文件发送完毕,这样就不会阻塞其他用户了。

服务器的例子:

复制

package nio.file;     import java.io.FileInputStream;     import java.io.IOException;     import java.net.InetSocketAddress;     import java.nio.ByteBuffer;     import java.nio.CharBuffer;      import java.nio.channels.FileChannel;     import java.nio.channels.SelectionKey;     import java.nio.channels.Selector;     import java.nio.channels.ServerSocketChannel;     import java.nio.channels.SocketChannel;     import java.nio.charset.Charset;     import java.nio.charset.CharsetDecoder;     import java.util.Iterator;      /**     * 测试文件下载的NIOServer     *     * @author tenyears.cn     */    public class NIOServer {      static int BLOCK = 4096;      // 处理与客户端的交互     public class HandleClient {     protected FileChannel channel;      protected ByteBuffer buffer;     public HandleClient() throws IOException {     this.channel = new FileInputStream(filename).getChannel();      this.buffer = ByteBuffer.allocate(BLOCK);      }      public ByteBuffer readBlock() {     try {     buffer.clear();      int count = channel.read(buffer);      buffer.flip();      if (count <= 0)      return null;      } catch (IOException e) {     e.printStackTrace();      }     return buffer;      }     public void close() {     try {     channel.close();     } catch (IOException e) {      e.printStackTrace();     }     }     }      protected Selector selector;      protected String filename = "d:\\bigfile.dat"; // a big file      protected ByteBuffer clientBuffer = ByteBuffer.allocate(BLOCK);     protected CharsetDecoder decoder;     public NIOServer(int port) throws IOException {     selector = this.getSelector(port);     Charset charset = Charset.forName("GB2312");      decoder = charset.newDecoder();      }     // 获取Selector     protected Selector getSelector(int port) throws IOException {     ServerSocketChannel server = ServerSocketChannel.open();     Selector sel = Selector.open();     server.socket().bind(new InetSocketAddress(port));     server.configureBlocking(false);     server.register(sel, SelectionKey.OP_ACCEPT);     return sel;      }      // 监听端口     public void listen() {     try {     for (;;) {     selector.select();     Iterator iter = selector.selectedKeys()     .iterator();     while (iter.hasNext()) {     SelectionKey key = iter.next();     iter.remove();     handleKey(key);     }     }     } catch (IOException e) {     e.printStackTrace();      }      }      // 处理事件      protected void handleKey(SelectionKey key) throws IOException {     if (key.isAcceptable()) { // 接收请求     ServerSocketChannel server = (ServerSocketChannel) key.channel();     SocketChannel channel = server.accept();     channel.configureBlocking(false);      channel.register(selector, SelectionKey.OP_READ);      } else if (key.isReadable()) { // 读信息      SocketChannel channel = (SocketChannel) key.channel();      int count = channel.read(clientBuffer);     if (count > 0) {     clientBuffer.flip();     CharBuffer charBuffer = decoder.decode(clientBuffer);     System.out.println("Client >>" + charBuffer.toString());      SelectionKey wKey = channel.register(selector,      SelectionKey.OP_WRITE);      wKey.attach(new HandleClient());      } else    channel.close();      clientBuffer.clear();     } else if (key.isWritable()) { // 写事件     SocketChannel channel = (SocketChannel) key.channel();     HandleClient handle = (HandleClient) key.attachment();      ByteBuffer block = handle.readBlock();      if (block != null)      channel.write(block);     else {     handle.close();     channel.close();      }     }      }     public static void main(String[] args) {     int port = 12345;     try {     NIOServer server = new NIOServer(port);     System.out.println("Listernint on " + port);     while (true) {     server.listen();     }     } catch (IOException e) {      e.printStackTrace();      }      }     }
  • 1.

  • 2.

  • 3.

  • 4.

  • 5.

  • 6.

  • 7.

  • 8.

  • 9.

  • 10.

  • 11.

  • 12.

  • 13.

  • 14.

  • 15.

  • 16.

  • 17.

  • 18.

  • 19.

  • 20.

  • 21.

  • 22.

  • 23.

  • 24.

  • 25.

  • 26.

  • 27.

  • 28.

  • 29.

  • 30.

  • 31.

  • 32.

  • 33.

  • 34.

  • 35.

  • 36.

  • 37.

  • 38.

  • 39.

  • 40.

  • 41.

  • 42.

  • 43.

  • 44.

  • 45.

  • 46.

  • 47.

  • 48.

  • 49.

  • 50.

  • 51.

  • 52.

  • 53.

  • 54.

  • 55.

  • 56.

  • 57.

  • 58.

  • 59.

  • 60.

  • 61.

  • 62.

  • 63.

  • 64.

  • 65.

  • 66.

  • 67.

  • 68.

  • 69.

  • 70.

  • 71.

  • 72.

  • 73.

  • 74.

  • 75.

  • 76.

  • 77.

  • 78.

  • 79.

  • 80.

  • 81.

  • 82.

  • 83.

  • 84.

  • 85.

  • 86.

  • 87.

  • 88.

  • 89.

  • 90.

  • 91.

  • 92.

  • 93.

  • 94.

  • 95.

  • 96.

  • 97.

  • 98.

  • 99.

  • 100.

  • 101.

  • 102.

  • 103.

  • 104.

  • 105.

  • 106.

  • 107.

  • 108.

  • 109.

  • 110.

  • 111.

  • 112.

  • 113.

  • 114.

  • 115.

  • 116.

  • 117.

  • 118.

  • 119.

  • 120.

  • 121.

  • 122.

  • 123.

  • 124.

  • 125.

  • 126.

  • 127.

  • 128.

  • 129.

该代码中,通过一个HandleClient来获取文件的一块数据,每一个客户都会分配一个HandleClient的实例。

下面是客户端请求的代码,也比较简单,模拟100个用户同时下载文件。

复制

package nio.file;     import java.io.IOException;     import java.net.InetSocketAddress;      import java.nio.ByteBuffer;      import java.nio.CharBuffer;      import java.nio.channels.SelectionKey;     import java.nio.channels.Selector;      import java.nio.channels.SocketChannel;      import java.nio.charset.Charset;      import java.nio.charset.CharsetEncoder;      import java.util.Iterator;     import java.util.concurrent.ExecutorService;      import java.util.concurrent.Executors;      /**      * 文件下载客户端      * @author tenyears.cn     */     public class NIOClient {      static int SIZE = 100;      static InetSocketAddress ip = new InetSocketAddress("localhost",12345);      static CharsetEncoder encoder = Charset.forName("GB2312").newEncoder();     static class Download implements Runnable {      protected int index;      public Download(int index) {      this.index = index;      }      public void run() {      try {      long start = System.currentTimeMillis();      SocketChannel client = SocketChannel.open();     client.configureBlocking(false);     Selector selector = Selector.open();      client.register(selector, SelectionKey.OP_CONNECT);      client.connect(ip);     ByteBuffer buffer = ByteBuffer.allocate(8 * 1024);     int total = 0;     FOR: for (;;) {      selector.select();      Iterator iter = selector.selectedKeys()      .iterator();      while (iter.hasNext()) {      SelectionKey key = iter.next();     iter.remove();     if (key.isConnectable()) {     SocketChannel channel = (SocketChannel) key     .channel();      if (channel.isConnectionPending())     channel.finishConnect();     channel.write(encoder.encode(CharBuffer     .wrap("Hello from " + index)));     channel.register(selector, SelectionKey.OP_READ);      } else if (key.isReadable()) {      SocketChannel channel = (SocketChannel) key      .channel();     int count = channel.read(buffer);     if (count > 0) {      total += count;      buffer.clear();     } else {      client.close();      break FOR;      }      }     }      }      double last = (System.currentTimeMillis() - start) * 1.0 / 1000;     System.out.println("Thread " + index + " downloaded " + total      + "bytes in " + last + "s.");      } catch (IOException e) {      e.printStackTrace();      }      }     }     public static void main(String[] args) throws IOException {     ExecutorService exec = Executors.newFixedThreadPool(SIZE);      for (int index = 0; index < SIZE; index++) {     exec.execute(new Download(index));      }      exec.shutdown();     }      }
  • 1.

  • 2.

  • 3.

  • 4.

  • 5.

  • 6.

  • 7.

  • 8.

  • 9.

  • 10.

  • 11.

  • 12.

  • 13.

  • 14.

  • 15.

  • 16.

  • 17.

  • 18.

  • 19.

  • 20.

  • 21.

  • 22.

  • 23.

  • 24.

  • 25.

  • 26.

  • 27.

  • 28.

  • 29.

  • 30.

  • 31.

  • 32.

  • 33.

  • 34.

  • 35.

  • 36.

  • 37.

  • 38.

  • 39.

  • 40.

  • 41.

  • 42.

  • 43.

  • 44.

  • 45.

  • 46.

  • 47.

  • 48.

  • 49.

  • 50.

  • 51.

  • 52.

  • 53.

  • 54.

  • 55.

  • 56.

  • 57.

  • 58.

  • 59.

  • 60.

  • 61.

  • 62.

  • 63.

  • 64.

  • 65.

  • 66.

  • 67.

  • 68.

  • 69.

  • 70.

  • 71.

  • 72.

  • 73.

  • 74.

  • 75.

  • 76.

  • 77.

  • 78.

  • 79.

  • 80.

  • 81.

操作系统的API epoll, select, NonBlocking的IO就是对select(Unix平台下)以及 WaitForMultipleObjects(Windows平台)的封装,是OS级别下的支持。

18    2011-03-11 09:51:47    Java NIO