Java NIO非阻塞服務(wù)器示例

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

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

以前一直用的是“ervery thread per connection”的服務(wù)器端模式，今天試了下NIO非阻塞模式的服務(wù)器。不過(guò)java不能實(shí)現(xiàn)I/O完成端口模型，這點(diǎn)很遺憾。

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

修改版本：

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;//默認(rèn)端口 
    private static int DEFAULT_BUFFERSIZE = 1024;//默認(rèn)緩沖區(qū)大小為1024字節(jié) 
    private static String DEFAULT_CHARSET = "GB2312";//默認(rèn)碼集 
    private static String DEFAULT_FILENAME = "bigfile.dat"; 
    private ServerSocketChannel channel; 
    private LinkedList<SocketChannel> clients; 
    private Selector selector;//選擇器 
    private ByteBuffer buffer;//字節(jié)緩沖區(qū) 
    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()  
         {//讀塊數(shù)據(jù) 
             return this.strGreeting; 
         } 
         public void close()  
         { 
              
         } 
    } 
 
    protected void handleKey(SelectionKey key) throws IOException 
    {//處理事件 
          if (key.isAcceptable())  
          { // 接收請(qǐng)求 
              ServerSocketChannel server = (ServerSocketChannel) key.channel();//取出對(duì)應(yīng)的服務(wù)器通道 
              SocketChannel channel = server.accept(); 
              channel.configureBlocking(false); 
              channel.register(selector, SelectionKey.OP_READ);//客戶socket通道注冊(cè)讀操作 
          } 
          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通道注冊(cè)寫操作 
                wKey.attach(new HandleClient()); 
              }  
              else 
              {//客戶已經(jīng)斷開 
                channel.close(); 
              } 
              this.buffer.clear();//清空緩沖區(qū) 
         } 
         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 
    { //服務(wù)器開始監(jiān)聽端口，提供服務(wù) 
        ServerSocket socket; 
        channel = ServerSocketChannel.open(); // 打開通道 
        socket = channel.socket();   //得到與通到相關(guān)的socket對(duì)象 
        socket.bind(new InetSocketAddress(port));   //將scoket榜定在制定的端口上 
        //配置通到使用非阻塞模式，在非阻塞模式下，可以編寫多道程序同時(shí)避免使用復(fù)雜的多線程 
        channel.configureBlocking(false);     
        channel.register(selector, SelectionKey.OP_ACCEPT); 
        try  
        { 
            while(true)  
            {//     與通常的程序不同，這里使用channel.accpet()接受客戶端連接請(qǐng)求，而不是在socket對(duì)象上調(diào)用accept(),這里在調(diào)用accept()方法時(shí)如果通道配置為非阻塞模式,那么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("服務(wù)器啟動(dòng)"); 
        SelectorServer server = new SelectorServer(SelectorServer.DEFAULT_SERVERPORT); 
        server.listen(); //服務(wù)器開始監(jiān)聽端口，提供服務(wù) 
    } 
 
}