使用Java NIO編寫高性能的服務器
從JDK 1.4開始,Java的標準庫中就包含了NIO,即所謂的“New IO”。其中最重要的功能就是提供了“非阻塞”的IO,當然包括了Socket。NonBlocking的IO就是對select(Unix平臺下)以及 WaitForMultipleObjects(Windows平臺)的封裝,提供了高性能、易伸縮的服務架構。
說來慚愧,直到JDK1.4才有這種功能,但遲到者不一定沒有螃蟹吃,NIO就提供了優(yōu)秀的面向對象的解決方案,可以很方便地編寫高性能的服務器。
話說回來,傳統(tǒng)的Server/Client實現(xiàn)是基于Thread per request,即服務器為每個客戶端請求建立一個線程處理,單獨負責處理一個客戶的請求。比如像Tomcat(新版本也會提供NIO方案)、Resin等Web服務器就是這樣實現(xiàn)的。當然為了減少瞬間峰值問題,服務器一般都使用線程池,規(guī)定了同時并發(fā)的最大數(shù)量,避免了線程的無限增長。
但這樣有一個問題:如果線程池的大小為100,當有100個用戶同時通過HTTP現(xiàn)在一個大文件時,服務器的線程池會用完,因為所有的線程都在傳輸大文件了,即使第101個請求者僅僅請求一個只有10字節(jié)的頁面,服務器也無法響應了,只有等到線程池中有空閑的線程出現(xiàn)。
另外,線程的開銷也是很大的,特別是達到了一個臨界值后,性能會顯著下降,這也限制了傳統(tǒng)的Socket方案無法應對并發(fā)量大的場合,而“非阻塞”的IO就能輕松解決這個問題。
下面只是一個簡單的例子:服務器提供了下載大型文件的功能,客戶端連接上服務器的12345端口后,就可以讀取服務器發(fā)送的文件內容信息了。注意這里的服務器只有一個主線程,沒有其他任何派生線程,讓我們看看NIO是如何用一個線程處理N個請求的。
NIO服務器最核心的一點就是反應器模式:當有感興趣的事件發(fā)生的,就通知對應的事件處理器去處理這個事件,如果沒有,則不處理。所以使用一個線程做輪詢就可以了。當然這里這是個例子,如果要獲得更高性能,可以使用少量的線程,一個負責接收請求,其他的負責處理請求,特別是對于多CPU時效率會更高。
關于使用NIO過程中出現(xiàn)的問題,最為普遍的就是為什么沒有請求時CPU的占用率為100%?出現(xiàn)這種問題的主要原因是注冊了不感興趣的事件,比如如果沒有數(shù)據(jù)要發(fā)到客戶端,而又注冊了寫事件(OP_WRITE),則在 Selector.select()上就會始終有事件出現(xiàn),CPU就一直處理了,而此時select()應該是阻塞的。
另外一個值得注意的問題是:由于只使用了一個線程(多個線程也如此)處理用戶請求,所以要避免線程被阻塞,解決方法是事件的處理者必須要即刻返回,不能陷入循環(huán)中,否則會影響其他用戶的請求速度。
具體到本例子中,由于文件比較大,如果一次性發(fā)送整個文件(這里的一次性不是指send整個文件內容,而是通過while循環(huán)不間斷的發(fā)送分組包),則主線程就會阻塞,其他用戶就不能響應了。這里的解決方法是當有WRITE事件時,僅僅是發(fā)送一個塊(比如4K字節(jié))。發(fā)完后,繼續(xù)等待WRITE事件出現(xiàn),依次處理,直到整個文件發(fā)送完畢,這樣就不會阻塞其他用戶了。
服務器的例子:
- 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;
 - }
 - // 監(jiān)聽端口
 - 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();
 - }
 - }
 - }
 
該代碼中,通過一個HandleClient來獲取文件的一塊數(shù)據(jù),每一個客戶都會分配一個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();
 - }
 - }
 
操作系統(tǒng)的API epoll, select, NonBlocking的IO就是對select(Unix平臺下)以及 WaitForMultipleObjects(Windows平臺)的封裝,是OS級別下的支持。
 
2011-12-15 13:28:57
 
 
 














 
 




 