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

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

Server:

/**   *    */  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;    /**   * @author ztz   *   */  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 = "XXX";   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 = (SelectionKey) 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();     }    }   }   /**    * @param args    */   public static void main(String[] args) {    try {     int port = 12345;     NIOServer server = new NIOServer(port);     System.out.println("Listening on "+port);     while(true){      server.listen();     }    } catch (IOException e) {     e.printStackTrace();    }       }    }

Client:

/**   *    */  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 ztz   *   */  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{        int index;    public DownLoad(int index){     this.index = index;    }    @Override    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 = (SelectionKey) 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+" download "+total+" bytes in "+last+"s.");             }     } catch (IOException e) {      e.printStackTrace();     }    }       }     /**    * @param args    */   public static void main(String[] args) {    ExecutorService exec = Executors.newFixedThreadPool(SIZE);    for(int index = 0;index < SIZE;index++){     exec.execute(new DownLoad(index));    }    exec.shutdown();   }    }