Java NIO服務(wù)器端開(kāi)發(fā)詳解

更新時(shí)間：2017年12月08日 15:30:11 作者：溫布利往事

這篇文章主要介紹了Java NIO服務(wù)器端開(kāi)發(fā)詳解，具有一定借鑒價(jià)值，需要的朋友可以參考下。

一、NIO類庫(kù)簡(jiǎn)介

　　1、緩沖區(qū)Buffer

　　Buffer是一個(gè)對(duì)象，包含一些要寫(xiě)入和讀出的數(shù)據(jù)。

　　在NIO中，所有的數(shù)據(jù)都是用緩沖區(qū)處理的，讀取數(shù)據(jù)時(shí)，它是從通道(Channel)直接讀到緩沖區(qū)中，在寫(xiě)入數(shù)據(jù)時(shí)，也是從緩沖區(qū)寫(xiě)入到通道。

　　緩沖區(qū)實(shí)質(zhì)上是一個(gè)數(shù)組，通常是一個(gè)字節(jié)數(shù)組(ByteBuffer)，也可以是其它類型的數(shù)組，此外緩沖區(qū)還提供了對(duì)數(shù)據(jù)的結(jié)構(gòu)化訪問(wèn)以及維護(hù)讀寫(xiě)位置等信息。

　　Buffer類的繼承關(guān)系如下圖所示：

2、通道Channel

　　Channel是一個(gè)通道，網(wǎng)絡(luò)數(shù)據(jù)通過(guò)Channel讀取和寫(xiě)入。通道和流的不同之處在于通道是雙向的(通道可以用于讀、寫(xiě)后者二者同時(shí)進(jìn)行)，流只是在一個(gè)方向上移動(dòng)。

　　Channel大體上可以分為兩類：用于網(wǎng)絡(luò)讀寫(xiě)的SelectableChannel(ServerSocketChannel和SocketChannel就是其子類)、用于文件操作的FileChannel。

　　下面的例子給出通過(guò)FileChannel來(lái)向文件中寫(xiě)入數(shù)據(jù)、從文件中讀取數(shù)據(jù)，將文件數(shù)據(jù)拷貝到另一個(gè)文件中：

public class NioTest
{
  public static void main(String[] args) throws IOException
  {
    copyFile();
  }
  //拷貝文件
  private static void copyFile()
  {
    FileInputStream in=null;
    FileOutputStream out=null;
    try
    {
      in=new FileInputStream("src/main/java/data/in-data.txt");
      out=new FileOutputStream("src/main/java/data/out-data.txt");
      FileChannel inChannel=in.getChannel();
      FileChannel outChannel=out.getChannel();
      ByteBuffer buffer=ByteBuffer.allocate(1024);
      int bytesRead = inChannel.read(buffer);
      while (bytesRead!=-1)
      {
        buffer.flip();
        outChannel.write(buffer);
        buffer.clear();
        bytesRead = inChannel.read(buffer);
      }
    }
    catch (FileNotFoundException e)
    {
      // TODO Auto-generated catch block
      e.printStackTrace();
    } catch (IOException e)
    {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }  
  }
  //寫(xiě)文件
  private static void writeFileNio()
  {
    try
    {
      RandomAccessFile fout = new RandomAccessFile("src/main/java/data/nio-data.txt", "rw");
      FileChannel fc=fout.getChannel();
      ByteBuffer buffer=ByteBuffer.allocate(1024);
      buffer.put("hi123".getBytes());
      buffer.flip();
      try
      {
        fc.write(buffer);
      } catch (IOException e)
      {
        // TODO Auto-generated catch block
        e.printStackTrace();
      }
    } 
    catch (FileNotFoundException e)
    {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
  }
  //讀文件
  private static void readFileNio()
  {
    FileInputStream fileInputStream;
    try
    {
      fileInputStream = new FileInputStream("src/main/java/data/nio-data.txt");
      FileChannel fileChannel=fileInputStream.getChannel();//從 FileInputStream 獲取通道
      ByteBuffer byteBuffer=ByteBuffer.allocate(1024);//創(chuàng)建緩沖區(qū)
      int bytesRead=fileChannel.read(byteBuffer);//將數(shù)據(jù)讀到緩沖區(qū)
      while(bytesRead!=-1)
      {
        /*limit=position
         * position=0;
         */
        byteBuffer.flip();
        //hasRemaining():告知在當(dāng)前位置和限制之間是否有元素
        while (byteBuffer.hasRemaining())
        {
          System.out.print((char) byteBuffer.get());
        }
        /*
         * 清空緩沖區(qū)
         * position=0;
         * limit=capacity;
         */
        byteBuffer.clear();
        bytesRead = fileChannel.read(byteBuffer);
      }
    } catch (FileNotFoundException e)
    {
      // TODO Auto-generated catch block
      e.printStackTrace();
    } catch (IOException e)
    {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
  }
}

3、多路復(fù)用器Selector

　　多路復(fù)用器提供選擇已經(jīng)就緒的任務(wù)的能力。Selector會(huì)不斷的輪詢注冊(cè)在其上的Channel，如果某個(gè)Channel上面發(fā)送讀或者寫(xiě)事件，這個(gè)Channel就處于就緒狀態(tài)，會(huì)被Selector輪詢出來(lái)，然后通過(guò)SelectionKey可以獲取就緒Channel的集合，進(jìn)行后續(xù)的I/O操作。

　　一個(gè)多路復(fù)用器Selector可以同時(shí)輪詢多個(gè)Channel，由于JDK使用了epoll代替了傳統(tǒng)的select實(shí)現(xiàn)，所以它沒(méi)有最大連接句柄1024/2048的限制，意味著只需要一個(gè)線程負(fù)責(zé)Selector的輪詢，就可以接入成千上萬(wàn)的客戶端。其模型如下圖所示：

用單線程處理一個(gè)Selector。要使用Selector，得向Selector注冊(cè)Channel，然后調(diào)用它的select()方法。這個(gè)方法會(huì)一直阻塞到某個(gè)注冊(cè)的通道有事件就緒。一旦這個(gè)方法返回，線程就可以處理這些事件，事件的例子有如新連接進(jìn)來(lái)，數(shù)據(jù)接收等。

　　注：

　　1、什么select模型？

　　select是事件觸發(fā)機(jī)制，當(dāng)?shù)却氖录l(fā)生就觸發(fā)進(jìn)行處理，多用于Linux實(shí)現(xiàn)的服務(wù)器對(duì)客戶端的處理。

　　可以阻塞地同時(shí)探測(cè)一組支持非阻塞的IO設(shè)備，是否有事件發(fā)生(如可讀、可寫(xiě)，有高優(yōu)先級(jí)錯(cuò)誤輸出等)，直至某一個(gè)設(shè)備觸發(fā)了事件或者超過(guò)了指定的等待時(shí)間。也就是它們的職責(zé)不是做IO，而是幫助調(diào)用者尋找當(dāng)前就緒的設(shè)備。

　　2、什么是epoll模型？

　　epoll的設(shè)計(jì)思路，是把select/poll單個(gè)的操作拆分為1個(gè)epoll_create+多個(gè)epoll_ctrl+一個(gè)wait。此外，內(nèi)核針對(duì)epoll操作添加了一個(gè)文件系統(tǒng)”eventpollfs”，每一個(gè)或者多個(gè)要監(jiān)視的文件描述符都有一個(gè)對(duì)應(yīng)的eventpollfs文件系統(tǒng)的inode節(jié)點(diǎn)，主要信息保存在eventpoll結(jié)構(gòu)體中。而被監(jiān)視的文件的重要信息則保存在epitem結(jié)構(gòu)體中。所以他們是一對(duì)多的關(guān)系。

二、NIO服務(wù)器端開(kāi)發(fā)

　　功能說(shuō)明：開(kāi)啟服務(wù)器端，對(duì)每一個(gè)接入的客戶端都向其發(fā)送hello字符串。

　　使用NIO進(jìn)行服務(wù)器端開(kāi)發(fā)主要有以下幾個(gè)步驟：

　　1、創(chuàng)建ServerSocketChannel，配置它為非阻塞模式

  serverSocketChannel = ServerSocketChannel.open();
  serverSocketChannel.configureBlocking(false);

　　2、綁定監(jiān)聽(tīng)，配置TCP參數(shù)，如backlog大小

  serverSocketChannel.socket().bind(new InetSocketAddress(8080));

　　3、創(chuàng)建一個(gè)獨(dú)立的I/O線程，用于輪詢多路復(fù)用器Selector

　　4、創(chuàng)建Selector，將之前創(chuàng)建的ServerSocketChannel注冊(cè)到Selector上，監(jiān)聽(tīng)SelectionKey.ACCEPT

　 selector=Selector.open();
  serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

　　5、啟動(dòng)I/O線程，在循環(huán)體內(nèi)執(zhí)行Selector.select()方法，輪詢就緒的Channel

while(true)
  {
    try
    {
      //select()阻塞到至少有一個(gè)通道在你注冊(cè)的事件上就緒了
      //如果沒(méi)有準(zhǔn)備好的channel,就在這一直阻塞
      //select(long timeout)和select()一樣，除了最長(zhǎng)會(huì)阻塞timeout毫秒(參數(shù))。
      selector.select();
    } 
    catch (IOException e)
    {
      // TODO Auto-generated catch block
      e.printStackTrace();
      break;
     }
　}

　　6、當(dāng)輪詢到了處于就緒狀態(tài)的Channel時(shí)，需對(duì)其進(jìn)行判斷，如果是OP_ACCEPT狀態(tài)，說(shuō)明是新的客戶端接入，則調(diào)用ServerSocketChannel.accept()方法接受新的客戶端

//返回已經(jīng)就緒的SelectionKey，然后迭代執(zhí)行
      Set<SelectionKey> readKeys=selector.selectedKeys();
      for(Iterator<SelectionKey> it=readKeys.iterator();it.hasNext();)
      {
        SelectionKey key=it.next();
        it.remove();
        try
        {
          if(key.isAcceptable())
          {
            ServerSocketChannel server=(ServerSocketChannel) key.channel();
            SocketChannel client=server.accept();
            client.configureBlocking(false);
            client.register(selector,SelectionKey.OP_WRITE);
          }
          else if(key.isWritable())
          {
            SocketChannel client=(SocketChannel) key.channel();
            ByteBuffer buffer=ByteBuffer.allocate(20);
            String str="hello";
            buffer=ByteBuffer.wrap(str.getBytes());
            client.write(buffer);
            key.cancel();
          } 
        }catch(IOException e)
        {
          e.printStackTrace();
          key.cancel();
          try
          {
            key.channel().close();
          } catch (IOException e1)
          {
            // TODO Auto-generated catch block
            e1.printStackTrace();
          }
          
        }
      }

　7、設(shè)置新接入的客戶端鏈路SocketChannel為非阻塞模式，配置其他的一些TCP參數(shù)

if(key.isAcceptable())
  {
    ServerSocketChannel server=(ServerSocketChannel) key.channel();
    SocketChannel client=server.accept();
    client.configureBlocking(false);
    ...
  }

　　8、將SocketChannel注冊(cè)到Selector，監(jiān)聽(tīng)OP_WRITE

　　client.register(selector,SelectionKey.OP_WRITE);

　　9、如果輪詢的Channel為OP_WRITE，則說(shuō)明要向SockChannel中寫(xiě)入數(shù)據(jù)，則構(gòu)造ByteBuffer對(duì)象，寫(xiě)入數(shù)據(jù)包

else if(key.isWritable())
  {
    SocketChannel client=(SocketChannel) key.channel();
    ByteBuffer buffer=ByteBuffer.allocate(20);
    String str="hello";
    buffer=ByteBuffer.wrap(str.getBytes());
    client.write(buffer);
    key.cancel();
  }

　完整代碼如下：

import java.io.IOException;
import java.net.InetSocketAddress;
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.Set;
public class ServerSocketChannelDemo
{
	public static void main(String[] args)
	  {
		ServerSocketChannel serverSocketChannel;
		Selector selector=null;
		try
		    {
			serverSocketChannel = ServerSocketChannel.open();
			serverSocketChannel.configureBlocking(false);
			serverSocketChannel.socket().bind(new InetSocketAddress(8080));
			selector=Selector.open();
			serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
		}
		catch (IOException e)
		    {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		while(true)
		    {
			try
			      {
				//select()阻塞到至少有一個(gè)通道在你注冊(cè)的事件上就緒了
				//如果沒(méi)有準(zhǔn)備好的channel,就在這一直阻塞
				//select(long timeout)和select()一樣，除了最長(zhǎng)會(huì)阻塞timeout毫秒(參數(shù))。
				selector.select();
			}
			catch (IOException e)
			      {
				// TODO Auto-generated catch block
				e.printStackTrace();
				break;
			}
			//返回已經(jīng)就緒的SelectionKey，然后迭代執(zhí)行
			Set<SelectionKey> readKeys=selector.selectedKeys();
			for (Iterator<SelectionKey> it=readKeys.iterator();it.hasNext();)
			      {
				SelectionKey key=it.next();
				it.remove();
				try
				        {
					if(key.isAcceptable())
					          {
						ServerSocketChannel server=(ServerSocketChannel) key.channel();
						SocketChannel client=server.accept();
						client.configureBlocking(false);
						client.register(selector,SelectionKey.OP_WRITE);
					} else if(key.isWritable())
					          {
						SocketChannel client=(SocketChannel) key.channel();
						ByteBuffer buffer=ByteBuffer.allocate(20);
						String str="hello";
						buffer=ByteBuffer.wrap(str.getBytes());
						client.write(buffer);
						key.cancel();
					}
				}
				catch(IOException e)
				        {
					e.printStackTrace();
					key.cancel();
					try
					          {
						key.channel().close();
					}
					catch (IOException e1)
					          {
						// TODO Auto-generated catch block
						e1.printStackTrace();
					}
				}
			}
		}
	}
}

我們用telnet localhost 8080模擬出多個(gè)客戶端：