example for NIO

NIO-Socket通讯,为我们解决了server端多线程设计方面的性能/吞吐量等多方面的问题,它提供了以非阻塞模式 + 线程池的方式来解决Server端高并发问题..NIO并不能显著的提升Client-server的通讯性能(其中包括全局性耗时总和,Server物理机资源开销和实际计算量),但是它可以确保Server端在支撑相应的并发量情况下,对物理资源的使用处于可控状态.对于开发者而言,NIO合理的使用了平台(OS/VM/Http协议)的特性并提供了高效的便捷的编程级别的API.

为了展示,NIO交互的基本特性,我们模拟了一个简单的场景:Client端向server端建立连接,并持续交付大量数据,Server负载client的数据传输和处理.此程序实例并没有太多的关注异常处理和业务性处理,也没有使用线程池作为server端socket句柄管理,不过你可以简单的修改代码也实现它.

1. TestMain.java:引导类
2. ClientControllor.java:client连接处理类,负责队列化数据提交,并负责维护socket句柄.
3. Packet.java:对于读取或者写入的buffer,进行二次封装,使其具有更好的可读性.
4. ServerControllor.java:server端连接处理类,负责接收连接和数据处理
5. ServerHandler.java:server端连接维护类.

TestMain.java:

Java代码  

1. package com.test.web;  
2.   
3.   
4. public class TestMain {  
5.   
6.     /** 
7.      * @param args 
8.      */  
9.     public static void main(String[] args) throws Exception{  
10.         int port = 30008;  
11.         ServerControllor sc = new ServerControllor(port);  
12.         sc.start();  
13.         Thread.sleep(2000);  
14.         ClientControllor cc = new ClientControllor("127.0.0.1", port);  
15.         cc.start();  
16.         Packet p1 = Packet.wrap("Hello,I am first!");  
17.         cc.put(p1);  
18.         Packet p2 = Packet.wrap("Hello,I am second!");  
19.         cc.put(p2);  
20.         Packet p3 = Packet.wrap("Hello,I am thread!");  
21.         cc.put(p3);  
22.   
23.     }  
24.   
25. }  

ClientControllor.java

Java代码  

1. package com.test.web;  
2.   
3. import java.net.InetSocketAddress;  
4. import java.net.SocketAddress;  
5. import java.nio.ByteBuffer;  
6. import java.nio.channels.SocketChannel;  
7. import java.util.concurrent.BlockingQueue;  
8. import java.util.concurrent.LinkedBlockingQueue;  
9. import java.util.zip.Adler32;  
10. import java.util.zip.Checksum;  
11.   
12. public class ClientControllor {  
13.   
14.     private BlockingQueue<Packet> inner = new LinkedBlockingQueue<Packet>(100);//no any more  
15.     private Object lock = new Object();  
16.     private InetSocketAddress remote;  
17.     private Thread thread = new ClientThread(remote);  
18.     public ClientControllor(String host,int port){  
19.         remote = new InetSocketAddress(host, port);  
20.     }  
21.       
22.     public void start(){  
23.         if(thread.isAlive() || remote == null){  
24.             return;  
25.         }  
26.         synchronized (lock) {  
27.             thread.start();  
28.         }  
29.               
30.           
31.     }  
32.     public boolean put(Packet packet){  
33.         return inner.offer(packet);  
34.     }  
35.       
36.     public void clear(){  
37.         inner.clear();  
38.     }  
39.       
40.     class ClientThread extends Thread {  
41.         SocketAddress remote;  
42.         SocketChannel channel;  
43.         ClientThread(SocketAddress remote){  
44.             this.remote = remote;  
45.         }  
46.         @Override  
47.         public void run(){  
48.             try{  
49.                 try{  
50.                     channel = SocketChannel.open();  
51.                     channel.configureBlocking(true);  
52.                     boolean isSuccess = channel.connect(new InetSocketAddress(30008));  
53.                     if(!isSuccess){  
54.                         while(!channel.finishConnect()){  
55.                             System.out.println("Client is connecting...");  
56.                         }  
57.                     }  
58.                     System.out.println("Client is connected.");  
59. //                  Selector selector = Selector.open();  
60. //                  channel.register(selector, SelectionKey.OP_WRITE);  
61. //                  while(selector.isOpen()){  
62. //                      selector.select();  
63. //                      Iterator<SelectionKey> it = selector.selectedKeys().iterator();  
64. //                      while(it.hasNext()){  
65. //                          SelectionKey key = it.next();  
66. //                          it.remove();  
67. //                          if(!key.isValid()){  
68. //                              continue;  
69. //                          }  
70. //                          if(key.isWritable()){  
71. //                              write();  
72. //                          }  
73. //                      }  
74. //                  }  
75.                     while(channel.isOpen()){  
76.                         write();  
77.                     }  
78.                 }catch(Exception e){  
79.                     e.printStackTrace();  
80.                 }finally{  
81.                     if(channel != null){  
82.                         try{  
83.                             channel.close();  
84.                         }catch(Exception ex){  
85.                             ex.printStackTrace();  
86.                         }  
87.                     }  
88.                 }  
89.             }catch(Exception e){  
90.                 e.printStackTrace();  
91.                 inner.clear();  
92.             }  
93.         }  
94.           
95.         private void write() throws Exception{  
96.             Packet packet = inner.take();  
97.             synchronized (lock) {  
98.                 ByteBuffer body = packet.getBuffer();//  
99.                 ByteBuffer head = ByteBuffer.allocate(4);  
100.                 head.putInt(body.limit());  
101.                 head.flip();  
102.                 while(head.hasRemaining()){  
103.                     channel.write(head);  
104.                 }  
105.                 Checksum checksum = new Adler32();  
106.                 while(body.hasRemaining()){  
107.                     checksum.update(body.get());  
108.                 }  
109.                 body.rewind();  
110.                 while(body.hasRemaining()){  
111.                     channel.write(body);  
112.                 }  
113.                 long cks = checksum.getValue();  
114.                 ByteBuffer tail = ByteBuffer.allocate(8);  
115.                 tail.putLong(cks);  
116.                 tail.flip();  
117.                 while(tail.hasRemaining()){  
118.                     channel.write(tail);  
119.                 }  
120.             }  
121.               
122.         }  
123.     }  
124. }  

Handler.java(接口,面向设计):

Java代码  

1. package com.test.web;  
2.   
3. import java.nio.channels.SocketChannel;  
4.   
5. public interface Handler {  
6.   
7.     public void handle(SocketChannel channel);  
8. }  

Packet.java

Java代码  

1. package com.test.web;  
2.   
3. import java.io.Serializable;  
4. import java.nio.ByteBuffer;  
5. import java.nio.charset.Charset;  
6.   
7. public class Packet implements Serializable {  
8.   
9.     /** 
10.      *  
11.      */  
12.     private static final long serialVersionUID = 7719389291885063462L;  
13.       
14.     private ByteBuffer buffer;  
15.       
16.     private static Charset charset = Charset.defaultCharset();  
17.       
18.     private Packet(ByteBuffer buffer){  
19.         this.buffer = buffer;  
20.     }  
21.       
22.     public String getDataAsString(){  
23.         return charset.decode(buffer).toString();  
24.     }  
25.       
26.     public byte[] getData(){  
27.         return buffer.array();  
28.     }  
29.       
30.     public ByteBuffer getBuffer(){  
31.         return this.buffer;  
32.     }  
33.       
34.       
35.     public static Packet wrap(ByteBuffer buffer){  
36.         return new Packet(buffer);  
37.     }  
38.       
39.     public static Packet wrap(String data){  
40.         ByteBuffer source = charset.encode(data);  
41.         return new Packet(source);  
42.     }  
43. }  

ServerControllor.java

Java代码  

1. package com.test.web;  
2.   
3. import java.net.InetSocketAddress;  
4. import java.nio.channels.SelectionKey;  
5. import java.nio.channels.Selector;  
6. import java.nio.channels.ServerSocketChannel;  
7. import java.nio.channels.SocketChannel;  
8. import java.util.Iterator;  
9.   
10. public class ServerControllor {  
11.     private int port;  
12.     private Thread thread = new ServerThread();;  
13.     private Object lock = new Object();  
14.     public ServerControllor(){  
15.         this(0);  
16.     }  
17.     public ServerControllor(int port){  
18.         this.port = port;  
19.     }  
20.       
21.     public void start(){  
22.         if(thread.isAlive()){  
23.             return;  
24.         }  
25.         synchronized (lock) {  
26.             thread.start();  
27.             System.out.println("Server starting....");  
28.         }  
29.     }  
30.       
31.       
32.     class ServerThread extends Thread {  
33.         private static final int TIMEOUT = 3000;  
34.         private ServerHandler handler = new ServerHandler();  
35.         @Override  
36.         public void run(){  
37.             try{  
38.                 ServerSocketChannel channel = null;  
39.                 try{  
40.                     channel = ServerSocketChannel.open();  
41.                     channel.configureBlocking(false);  
42.                     channel.socket().setReuseAddress(true);  
43.                     channel.socket().bind(new InetSocketAddress(port));  
44.                     Selector selector = Selector.open();  
45.                     channel.register(selector, SelectionKey.OP_ACCEPT);  
46.                     while(selector.isOpen()){  
47.                         System.out.println("Server is running,port:" + channel.socket().getLocalPort());  
48.                         if(selector.select(TIMEOUT) == 0){  
49.                             continue;  
50.                         }  
51.                         Iterator<SelectionKey> it = selector.selectedKeys().iterator();  
52.                         while(it.hasNext()){  
53.                             SelectionKey key = it.next();  
54.                             it.remove();  
55.                             if(!key.isValid()){  
56.                                 continue;  
57.                             }  
58.                             if(key.isAcceptable()){  
59.                                 accept(key);  
60.                             }else if(key.isReadable()){  
61.                                 read(key);  
62.                             }  
63.                         }  
64.                     }  
65.                 }catch(Exception e){  
66.                     e.printStackTrace();  
67.                 }finally{  
68.                     if(channel != null){  
69.                         try{  
70.                             channel.close();  
71.                         }catch(Exception ex){  
72.                             ex.printStackTrace();  
73.                         }  
74.                     }  
75.                 }  
76.             }catch(Exception e){  
77.                 e.printStackTrace();  
78.             }  
79.         }  
80.           
81.         private void accept(SelectionKey key) throws Exception{  
82.             SocketChannel socketChannel = ((ServerSocketChannel) key.channel()).accept();  
83.             socketChannel.configureBlocking(true);  
84.             //socketChannel.register(key.selector(), SelectionKey.OP_READ);  
85.             handler.handle(socketChannel);  
86.         }  
87.           
88.         private void read(SelectionKey key) throws Exception{  
89.             SocketChannel channel = (SocketChannel)key.channel();  
90.             //handler.handle(channel);  
91.         }  
92.     }  
93. }  

ServerHandler.java

Java代码  

1. package com.test.web;  
2.   
3. import java.nio.ByteBuffer;  
4. import java.nio.channels.SocketChannel;  
5. import java.util.HashMap;  
6. import java.util.Map;  
7. import java.util.concurrent.Semaphore;  
8. import java.util.zip.Adler32;  
9. import java.util.zip.Checksum;  
10.   
11. class ServerHandler implements Handler {  
12.   
13.     private static Semaphore semaphore = new Semaphore(Runtime.getRuntime().availableProcessors() + 1);  
14.       
15.     private static Map<SocketChannel,Thread> holder = new HashMap<SocketChannel,Thread>(32);  
16.       
17.     @Override  
18.     public void handle(SocketChannel channel) {  
19.         synchronized (holder) {  
20.             if(holder.containsKey(channel)){  
21.                 return;  
22.             }  
23.             Thread t = new ReadThread(channel);  
24.             holder.put(channel, t);  
25.             t.start();  
26.         }  
27.     }  
28.       
29.       
30.     static class ReadThread extends Thread{  
31.         SocketChannel channel;  
32.         ReadThread(SocketChannel channel){  
33.             this.channel = channel;  
34.         }  
35.         @Override  
36.         public void run(){  
37.             try{  
38.                 semaphore.acquire();  
39.                 boolean eof = false;  
40.                 while(channel.isOpen()){  
41.                     //ByteBuffer byteBuffer = new ByteBuffer(1024);  
42.                     ByteBuffer head = ByteBuffer.allocate(4);//int for data-size  
43.                     while(true){  
44.                         int cb = channel.read(head);  
45.                         if(cb == -1){  
46.                             throw new RuntimeException("EOF error,data lost!");  
47.                         }  
48.                         if(isFull(head)){  
49.                             break;  
50.                         }  
51.                     }  
52.                     head.flip();  
53.                     int dataSize = head.getInt();  
54.                     if(dataSize <= 0){  
55.                         throw new RuntimeException("Data format error,something lost???");  
56.                     }  
57.                     ByteBuffer body = ByteBuffer.allocate(dataSize);  
58.                     while(true){  
59.                         int cb = channel.read(body);  
60.                         if(cb == -1){  
61.                             throw new RuntimeException("EOF error,data lost!");  
62.                         }else if(cb == 0 && this.isFull(body)){  
63.                             break;  
64.                         }  
65.                     }  
66.                     ByteBuffer tail = ByteBuffer.allocate(8);//int for data-size  
67.                     while(true){  
68.                         int cb = channel.read(tail);  
69.                         if(cb == -1){  
70.                             eof = true;  
71.                         }  
72.                         if(isFull(tail)){  
73.                             break;  
74.                         }  
75.                     }  
76.                     tail.flip();  
77.                     long sck = tail.getLong();  
78.                     Checksum checksum = new Adler32();  
79.                     checksum.update(body.array(), 0, dataSize);  
80.                     long cck = checksum.getValue();  
81.                     if(sck != cck){  
82.                         throw new RuntimeException("Sorry,some data lost or be modified,please check!");  
83.                     }  
84.                     body.flip();  
85.                     Packet packet = Packet.wrap(body);  
86.                     System.out.println(packet.getDataAsString());  
87.                     if(eof){  
88.                         break;  
89.                     }  
90.                 }  
91.             }catch(Exception e){  
92.                 e.printStackTrace();  
93.             }finally{  
94.                 if(channel != null){  
95.                     try{  
96.                         channel.close();  
97.                     }catch(Exception ex){  
98.                         ex.printStackTrace();  
99.                     }  
100.                 }  
101.                 holder.remove(channel);  
102.                 semaphore.release();  
103.             }  
104.         }  
105.           
106.         private boolean isFull(ByteBuffer byteBuffer){  
107.             return byteBuffer.position() == byteBuffer.capacity() ? true : false;  
108.         }  
109.     }  
110.   
111. }  

 原文链接：[http://wely.iteye.com/blog/2227865]