Java NIO应用实例

Posted 2020-08-15 WindWant

同步非阻塞IO （NIO）

• NIO是基于事件驱动思想的，实现上通常采用Reactor（http://en.wikipedia.org/wiki/Reactor_pattern）模式，从程序角度而言，当发起IO的读或写操作时，是非阻塞的；当socket有流可读或可写入socket时，操作系统会相应的通知引用程序进行处理，应用再将流读取到缓冲区或写入操作系统。
• 对于网络IO而言，主要有连接建立、流读取及流写入三种事件、linux2.6以后的版本使用epoll（http://lse.sourceforge.net/epoll/index.html）方式实现NIO。
• select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。它的基本原理就是select/epoll这个function会不断的轮询所负责的所有socket，当某个socket有数据到达了，就通知用户进程。
• 当用户进程调用了select，那么整个进程会被block，而同时，kernel会“监视”所有select负责的socket，当任何一个socket中的数据准备好了，select就会返回。这个时候用户进程再调用read操作，将数据从kernel拷贝到用户进程。用select的优势在于它可以同时处理多个connection。

server：

  1 package org.windwant.nio;
  2 
  3 import java.io.IOException;
  4 import java.net.InetSocketAddress;
  5 import java.net.ServerSocket;
  6 import java.nio.ByteBuffer;
  7 import java.nio.channels.SelectionKey;
  8 import java.nio.channels.Selector;
  9 import java.nio.channels.ServerSocketChannel;
 10 import java.nio.channels.SocketChannel;
 11 import java.util.Iterator;
 12 import java.util.Set;
 13 
 14 /**
 15  * ServerSocketChannel
 16  */
 17 public class Nioserver {
 18     /*标识数字*/
 19     private int flag = 0;
 20     /*缓冲区大小*/
 21     private int BLOCK = 2048;
 22     /*接受数据缓冲区*/
 23     private ByteBuffer sendbuffer = ByteBuffer.allocate(BLOCK);
 24     /*发送数据缓冲区*/
 25     private ByteBuffer receivebuffer = ByteBuffer.allocate(BLOCK);
 26     private Selector selector;
 27 
 28     public NIOServer(int port) throws IOException {
 29         // 打开服务器套接字通道
 30 
 31         ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
 32         // 服务器配置为非阻塞
 33 
 34         serverSocketChannel.configureBlocking(false);
 35         // 检索与此通道关联的服务器套接字
 36 
 37         ServerSocket serverSocket = serverSocketChannel.socket();
 38         // 进行服务的绑定
 39 
 40         serverSocket.bind(new InetSocketAddress(port));
 41         // 通过open()方法找到Selector
 42 
 43         selector = Selector.open();
 44         // 注册到selector，等待连接
 45 
 46         serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
 47         System.out.println("Server Start----8888:");
 48     }
 49 
 50 
 51     // 监听
 52 
 53     private void listen() throws IOException {
 54         while (true) {
 55             // 选择一组键，并且相应的通道已经打开
 56 
 57             selector.select();
 58             // 返回此选择器的已选择键集。
 59 
 60             Set<SelectionKey> selectionKeys = selector.selectedKeys();
 61             Iterator<SelectionKey> iterator = selectionKeys.iterator();
 62             while (iterator.hasNext()) {
 63                 SelectionKey selectionKey = iterator.next();
 64                 iterator.remove();
 65                 handleKey(selectionKey);
 66             }
 67         }
 68     }
 69 
 70     // 处理请求
 71 
 72     private void handleKey(SelectionKey selectionKey) throws IOException {
 73         // 接受请求
 74 
 75         ServerSocketChannel server = null;
 76         SocketChannel client = null;
 77         String receiveText;
 78         String sendText;
 79         int count=0;
 80         // 测试此键的通道是否已准备好接受新的套接字连接。
 81 
 82         if (selectionKey.isAcceptable()) {
 83             // 返回为之创建此键的通道。
 84 
 85             server = (ServerSocketChannel) selectionKey.channel();
 86             // 接受到此通道套接字的连接。
 87 
 88             // 此方法返回的套接字通道（如果有）将处于阻塞模式。
 89 
 90             client = server.accept();
 91             // 配置为非阻塞
 92 
 93             client.configureBlocking(false);
 94             // 注册到selector，等待连接
 95 
 96             client.register(selector, SelectionKey.OP_READ);
 97         } else if (selectionKey.isReadable()) {
 98             // 返回为之创建此键的通道。
 99 
100             client = (SocketChannel) selectionKey.channel();
101             //将缓冲区清空以备下次读取
102 
103             receivebuffer.clear();
104             //读取服务器发送来的数据到缓冲区中
105 
106             count = client.read(receivebuffer);
107             if (count > 0) {
108                 receiveText = new String( receivebuffer.array(),0,count);
109                 System.out.println("服务器端接受客户端数据--:"+receiveText);
110                 client.register(selector, SelectionKey.OP_WRITE);
111             }
112         } else if (selectionKey.isWritable()) {
113             //将缓冲区清空以备下次写入
114 
115             sendbuffer.clear();
116             // 返回为之创建此键的通道。
117 
118             client = (SocketChannel) selectionKey.channel();
119 
120 
121             sendText = "<h1>message from server: this is the test message!</h1>";
122 
123 //            sendText="message from server--" + flag++;
124             //向缓冲区中输入数据
125 
126             sendbuffer.put(sendText.getBytes());
127             //将缓冲区各标志复位,因为向里面put了数据标志被改变要想从中读取数据发向服务器,就要复位
128 
129             sendbuffer.flip();
130             //输出到通道
131 
132             client.write(sendbuffer);
133             System.out.println("服务器端向客户端发送数据--："+sendText);
134             client.register(selector, SelectionKey.OP_READ);
135         }
136     }
137 
138     /**
139      * @param args
140      * @throws IOException
141      */
142     public static void main(String[] args) throws IOException {
143         int port = 8888;
144         NIOServer server = new NIOServer(port);
145         server.listen();
146     }
147 }

client：

  1 package org.windwant.nio;
  2 
  3 import java.io.IOException;
  4 import java.net.InetSocketAddress;
  5 import java.nio.ByteBuffer;
  6 import java.nio.channels.SelectionKey;
  7 import java.nio.channels.Selector;
  8 import java.nio.channels.SocketChannel;
  9 import java.util.Iterator;
 10 import java.util.Set;
 11 
 12 /**
 13  * SocketChannel
 14  */
 15 public class NIOClient {
 16     /*标识数字*/
 17     private static int flag = 0;
 18     /*缓冲区大小*/
 19     private static int BLOCK = 4096;
 20     /*接受数据缓冲区*/
 21     private static ByteBuffer sendbuffer = ByteBuffer.allocate(BLOCK);
 22     /*发送数据缓冲区*/
 23     private static ByteBuffer receivebuffer = ByteBuffer.allocate(BLOCK);
 24     /*服务器端地址*/
 25     private final static InetSocketAddress SERVER_ADDRESS = new InetSocketAddress(
 26             "localhost", 8888);
 27 
 28     public static void main(String[] args) throws IOException {
 29         // 打开socket通道
 30 
 31         SocketChannel socketChannel = SocketChannel.open();
 32         // 设置为非阻塞方式
 33 
 34         socketChannel.configureBlocking(false);
 35         // 打开选择器
 36 
 37         Selector selector = Selector.open();
 38         // 注册连接服务端socket动作
 39 
 40         socketChannel.register(selector, SelectionKey.OP_CONNECT);
 41         // 连接
 42 
 43         socketChannel.connect(SERVER_ADDRESS);
 44         // 分配缓冲区大小内存
 45 
 46 
 47         Set<SelectionKey> selectionKeys;
 48         Iterator<SelectionKey> iterator;
 49         SelectionKey selectionKey;
 50         SocketChannel client;
 51         String receiveText;
 52         String sendText;
 53         int count=0;
 54 
 55         while (true) {
 56             //选择一组键，其相应的通道已为 I/O 操作准备就绪。
 57 
 58             //此方法执行处于阻塞模式的选择操作。
 59 
 60             selector.select();
 61             //返回此选择器的已选择键集。
 62 
 63             selectionKeys = selector.selectedKeys();
 64             //System.out.println(selectionKeys.size());
 65 
 66             iterator = selectionKeys.iterator();
 67             while (iterator.hasNext()) {
 68                 selectionKey = iterator.next();
 69                 if (selectionKey.isConnectable()) {
 70                     System.out.println("client connect");
 71                     client = (SocketChannel) selectionKey.channel();
 72                     // 判断此通道上是否正在进行连接操作。
 73 
 74                     // 完成套接字通道的连接过程。
 75 
 76                     if (client.isConnectionPending()) {
 77                         client.finishConnect();
 78                         System.out.println("完成连接!");
 79                         sendbuffer.clear();
 80                         sendbuffer.put("Hello,Server".getBytes());
 81                         sendbuffer.flip();
 82                         client.write(sendbuffer);
 83                     }
 84                     client.register(selector, SelectionKey.OP_READ);
 85                 } else if (selectionKey.isReadable()) {
 86                     client = (SocketChannel) selectionKey.channel();
 87                     //将缓冲区清空以备下次读取
 88 
 89                     receivebuffer.clear();
 90                     //读取服务器发送来的数据到缓冲区中
 91 
 92                     count=client.read(receivebuffer);
 93                     if(count>0){
 94                         receiveText = new String( receivebuffer.array(),0,count);
 95                         System.out.println("客户端接受服务器端数据--:"+receiveText);
 96                         client.register(selector, SelectionKey.OP_WRITE);
 97                     }
 98 
 99                 } else if (selectionKey.isWritable()) {
100                     sendbuffer.clear();
101                     client = (SocketChannel) selectionKey.channel();
102                     sendText = "message from client--" + (flag++);
103                     sendbuffer.put(sendText.getBytes());
104                     //将缓冲区各标志复位,因为向里面put了数据标志被改变要想从中读取数据发向服务器,就要复位
105 
106                     sendbuffer.flip();
107                     client.write(sendbuffer);
108                     System.out.println("客户端向服务器端发送数据--："+sendText);
109                     client.register(selector, SelectionKey.OP_READ);
110                 }
111             }
112             selectionKeys.clear();
113         }
114     }
115 }

项目地址：https://github.com/windwant/nio-test