Java NIO非阻塞服務器示例

作者：洞庭散人 2011-12-08 10:12:34

以前一直用的是“ervery thread per connection”的服務器端模式，今天試了下NIO非阻塞模式的服務器。

以前一直用的是“ervery thread per connection”的服務器端模式，今天試了下NIO非阻塞模式的服務器。不過java不能實現I/O完成端口模型，這點很遺憾。

package com.vista.Server; 
 
import java.io.IOException; 
import java.net.InetSocketAddress; 
import java.net.ServerSocket; 
import java.nio.ByteBuffer; 
import java.nio.channels.SelectionKey; 
import java.nio.channels.Selector; 
import java.nio.channels.ServerSocketChannel; 
import java.nio.channels.SocketChannel; 
import java.util.Iterator; 
import java.util.LinkedList; 
import java.util.Set; 
 
public class SelectorServer  
{ 
    private static int DEFAULT_SERVERPORT = 6018;//默認端口 
    private static int DEFAULT_BUFFERSIZE = 1024;//默認緩沖區大小為1024字節 
    private ServerSocketChannel channel; 
    private LinkedList<SocketChannel> clients; 
    private Selector readSelector; 
    private ByteBuffer buffer;//字節緩沖區 
    private int port; 
     
    public SelectorServer(int port) throws IOException 
    { 
        this.port = port; 
        this.clients = new LinkedList<SocketChannel>(); 
        this.channel = null; 
        this.readSelector = Selector.open();//打開選擇器 
        this.buffer = ByteBuffer.allocate(DEFAULT_BUFFERSIZE); 
    } 
     // 服務器程序在服務循環中調用sericeClients()方法為已接受的客戶服務 
    public void serviceClients()throws IOException 
    { 
        Set keys; 
        Iterator it; 
        SelectionKey key; 
        SocketChannel client; 
        // 在readSelector上調用select()方法，參數1代表如果調用select的時候 那么阻塞最多1秒鐘等待可用的客戶端連接 
        if(readSelector.select(1) > 0) 
        { 
            keys = readSelector.selectedKeys(); // 取得代表端通道的鍵集合 
            it = keys.iterator(); 
           // 遍歷，為每一個客戶服務  
            while(it.hasNext())  
            { 
               key = (SelectionKey)it.next(); 
               if(key.isReadable()) 
               { // 如果通道可讀，那么讀此通道到buffer中 
                  int bytes; 
                  client = (SocketChannel)key.channel();// 取得鍵對應的通道 
                  buffer.clear(); // 清空緩沖區中的內容，設置好position,limit，準備接受數據 
                  bytes = client.read(buffer); // 從通道中讀數據到緩沖中，返回讀取得字節數 
                  if(bytes >= 0)  
                  { 
                     buffer.flip(); // 準備將緩沖中的數據寫回到通道中 
                     client.write(buffer);  // 數據寫回到通道中 
                  }  
                  else if(bytes < 0)  
                  { // 如果返回小于零的值代表讀到了流的末尾 
                     clients.remove(client); 
                  // 通道關閉時，選擇鍵也被取消 
                     client.close(); 
                  } 
               } 
            } 
         } 
    } 
     
    public void registerClient(SocketChannel client) throws IOException 
    {// 配置和注冊代表客戶連接的通道對象 
        client.configureBlocking(false);  // 設置此通道使用非阻塞模式     
        client.register(readSelector, SelectionKey.OP_READ); // 將這個通道注冊到選擇器上 
        clients.add(client); //保存這個通道對象 
    } 
    public void listen() throws IOException 
    { //服務器開始監聽端口，提供服務 
        ServerSocket socket; 
        SocketChannel client; 
        channel = ServerSocketChannel.open(); // 打開通道 
        socket = channel.socket();   //得到與通到相關的socket對象 
        socket.bind(new InetSocketAddress(port), 10);   //將scoket榜定在制定的端口上 
        //配置通到使用非阻塞模式，在非阻塞模式下，可以編寫多道程序同時避免使用復雜的多線程 
        channel.configureBlocking(false);     
        try  
        { 
            while(true)  
            {//     與通常的程序不同，這里使用channel.accpet()接受客戶端連接請求，而不是在socket對象上調用accept(),這里在調用accept()方法時如果通道配置為非阻塞模式,那么accept()方法立即返回null，并不阻塞 
                client = channel.accept();     
                if(client != null) 
                { 
                    registerClient(client); // 注冊客戶信息 
                } 
                serviceClients();  // 為以連接的客戶服務 
            } 
        }  
        finally  
        { 
            socket.close(); // 關閉socket，關閉socket會同時關閉與此socket關聯的通道 
        } 
    } 
    public static void main(String[] args) throws IOException  
    { 
        System.out.println("服務器啟動"); 
        SelectorServer server = new SelectorServer(SelectorServer.DEFAULT_SERVERPORT); 
        server.listen(); //服務器開始監聽端口，提供服務 
 
         
    } 
 
}

修改版本：

package com.vista.Server; 
 
import java.io.BufferedWriter; 
import java.io.FileInputStream; 
import java.io.IOException; 
import java.io.OutputStreamWriter; 
import java.io.PrintWriter; 
import java.net.InetSocketAddress; 
import java.net.ServerSocket; 
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; 
import java.util.LinkedList; 
import java.util.Set; 
 
public class SelectorServer  
{ 
    private static int DEFAULT_SERVERPORT = 6018;//默認端口 
    private static int DEFAULT_BUFFERSIZE = 1024;//默認緩沖區大小為1024字節 
    private static String DEFAULT_CHARSET = "GB2312";//默認碼集 
    private static String DEFAULT_FILENAME = "bigfile.dat"; 
    private ServerSocketChannel channel; 
    private LinkedList<SocketChannel> clients; 
    private Selector selector;//選擇器 
    private ByteBuffer buffer;//字節緩沖區 
    private int port; 
    private Charset charset;//字符集 
    private CharsetDecoder decoder;//解碼器 
     
     
    public SelectorServer(int port) throws IOException 
    { 
        this.port = port; 
        this.clients = new LinkedList<SocketChannel>(); 
        this.channel = null; 
        this.selector = Selector.open();//打開選擇器 
        this.buffer = ByteBuffer.allocate(DEFAULT_BUFFERSIZE); 
        this.charset = Charset.forName(DEFAULT_CHARSET); 
        this.decoder = this.charset.newDecoder(); 
         
    } 
     
     private class HandleClient  
     { 
         private String strGreeting = "welcome to VistaQQ"; 
         public HandleClient() throws IOException  
         { 
         } 
         public String readBlock()  
         {//讀塊數據 
             return this.strGreeting; 
         } 
         public void close()  
         { 
              
         } 
    } 
 
    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);//客戶socket通道注冊讀操作 
          } 
          else if (key.isReadable())  
          { // 讀信息 
              SocketChannel channel = (SocketChannel) key.channel(); 
              int count = channel.read(this.buffer); 
              if (count > 0)  
              { 
                this.buffer.flip(); 
                CharBuffer charBuffer = decoder.decode(this.buffer); 
                System.out.println("Client >>" + charBuffer.toString()); 
                SelectionKey wKey = channel.register(selector, 
                    SelectionKey.OP_WRITE);//為客戶sockt通道注冊寫操作 
                wKey.attach(new HandleClient()); 
              }  
              else 
              {//客戶已經斷開 
                channel.close(); 
              } 
              this.buffer.clear();//清空緩沖區 
         } 
         else if (key.isWritable())  
         { // 寫事件 
              SocketChannel channel = (SocketChannel) key.channel(); 
              HandleClient handle = (HandleClient) key.attachment();//取出處理者 
              ByteBuffer block = ByteBuffer.wrap(handle.readBlock().getBytes()); 
              channel.write(block); 
             // channel.socket().getInputStream().(block); 
//              PrintWriter out = new PrintWriter(new BufferedWriter(new OutputStreamWriter( 
//                        channel.socket().getOutputStream())), true); 
//              out.write(block.toString()); 
 
        } 
 
    } 
    public void listen() throws IOException 
    { //服務器開始監聽端口，提供服務 
        ServerSocket socket; 
        channel = ServerSocketChannel.open(); // 打開通道 
        socket = channel.socket();   //得到與通到相關的socket對象 
        socket.bind(new InetSocketAddress(port));   //將scoket榜定在制定的端口上 
        //配置通到使用非阻塞模式，在非阻塞模式下，可以編寫多道程序同時避免使用復雜的多線程 
        channel.configureBlocking(false);     
        channel.register(selector, SelectionKey.OP_ACCEPT); 
        try  
        { 
            while(true)  
            {//     與通常的程序不同，這里使用channel.accpet()接受客戶端連接請求，而不是在socket對象上調用accept(),這里在調用accept()方法時如果通道配置為非阻塞模式,那么accept()方法立即返回null，并不阻塞 
                this.selector.select(); 
                Iterator iter = this.selector.selectedKeys().iterator(); 
                while(iter.hasNext()) 
                { 
                    SelectionKey key = (SelectionKey)iter.next(); 
                    iter.remove(); 
                    this.handleKey(key); 
                     
                } 
            } 
        }  
        catch(IOException ex) 
        { 
            ex.printStackTrace(); 
        } 
    } 
    public static void main(String[] args) throws IOException  
    { 
        System.out.println("服務器啟動"); 
        SelectorServer server = new SelectorServer(SelectorServer.DEFAULT_SERVERPORT); 
        server.listen(); //服務器開始監聽端口，提供服務 
    } 
 
}

原文鏈接：http://www.cnblogs.com/phinecos/archive/2008/07/17/1245428.html

【編輯推薦】

基于Java NIO的即時聊天服務器模型
Java解讀NIO Socket非阻塞模式
利用NIO建立Socket服務器
用Java.nio.* 進行網絡編程
Java NIO的wakeup剖析

責任編輯：林師授來源：洞庭散人的博客

Java NIO