BIO,NIO区别
Posted fjax
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了BIO,NIO区别相关的知识，希望对你有一定的参考价值。
1、BIO编程 1.1、传统的BIO编程 网络编程的基本模型是C/S模型，即两个进程间的通信。
服务端提供IP和监听端口，客户端通过连接操作想服务端监听的地址发起连接请求，通过三次握手连接，如果连接成功建立，双方就可以通过套接字进行通信。
传统的同步阻塞模型开发中，ServerSocket负责绑定IP地址，启动监听端口；Socket负责发起连接操作。连接成功后，双方通过输入和输出流进行同步阻塞式通信。
简单的描述一下BIO的服务端通信模型：采用BIO通信模型的服务端，通常由一个独立的Acceptor线程负责监听客户端的连接，它接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理没处理完成后，通过输出流返回应答给客户端，线程销毁。即典型的一请求一应答通宵模型。
传统BIO通信模型图：
该模型最大的问题就是缺乏弹性伸缩能力，当客户端并发访问量增加后，服务端的线程个数和客户端并发访问数呈1:1的正比关系，Java中的线程也是比较宝贵的系统资源，线程数量快速膨胀后，系统的性能将急剧下降，随着访问量的继续增大，系统最终就死-掉-了。
同步阻塞式I/O创建的Server源码：
1 package com.anxpp.io.calculator.bio; 2 import java.io.IOException; 3 import java.net.ServerSocket; 4 import java.net.Socket; 5 /** 6 * BIO服务端源码 7 * @author yangtao__anxpp.com 8 * @version 1.0 9 */ 10 public final class ServerNormal { 11 //默认的端口号 12 private static int DEFAULT_PORT = 12345; 13 //单例的ServerSocket 14 private static ServerSocket server; 15 //根据传入参数设置监听端口，如果没有参数调用以下方法并使用默认值 16 public static void start() throws IOException{ 17 //使用默认值 18 start(DEFAULT_PORT); 19 } 20 //这个方法不会被大量并发访问，不太需要考虑效率，直接进行方法同步就行了 21 public synchronized static void start(int port) throws IOException{ 22 if(server != null) return; 23 try{ 24 //通过构造函数创建ServerSocket 25 //如果端口合法且空闲，服务端就监听成功 26 server = new ServerSocket(port); 27 System.out.println("服务器已启动，端口号：" + port); 28 //通过无线循环监听客户端连接 29 //如果没有客户端接入，将阻塞在accept操作上。 30 while(true){ 31 Socket socket = server.accept(); 32 //当有新的客户端接入时，会执行下面的代码 33 //然后创建一个新的线程处理这条Socket链路 34 new Thread(new ServerHandler(socket)).start(); 35 } 36 }finally{ 37 //一些必要的清理工作 38 if(server != null){ 39 System.out.println("服务器已关闭。"); 40 server.close(); 41 server = null; 42 } 43 } 44 } 45 }
客户端消息处理线程ServerHandler源码：
1 package com.anxpp.io.calculator.bio; 2 import java.io.BufferedReader; 3 import java.io.IOException; 4 import java.io.InputStreamReader; 5 import java.io.PrintWriter; 6 import java.net.Socket; 7 8 import com.anxpp.io.utils.Calculator; 9 /** 10 * 客户端线程 11 * @author yangtao__anxpp.com 12 * 用于处理一个客户端的Socket链路 13 */ 14 public class ServerHandler implements Runnable{ 15 private Socket socket; 16 public ServerHandler(Socket socket) { 17 this.socket = socket; 18 } 19 @Override 20 public void run() { 21 BufferedReader in = null; 22 PrintWriter out = null; 23 try{ 24 in = new BufferedReader(new InputStreamReader(socket.getInputStream())); 25 out = new PrintWriter(socket.getOutputStream(),true); 26 String expression; 27 String result; 28 while(true){ 29 //通过BufferedReader读取一行 30 //如果已经读到输入流尾部，返回null,退出循环 31 //如果得到非空值，就尝试计算结果并返回 32 if((expression = in.readLine())==null) break; 33 System.out.println("服务器收到消息：" + expression); 34 try{ 35 result = Calculator.cal(expression).toString(); 36 }catch(Exception e){ 37 result = "计算错误：" + e.getMessage(); 38 } 39 out.println(result); 40 } 41 }catch(Exception e){ 42 e.printStackTrace(); 43 }finally{ 44 //一些必要的清理工作 45 if(in != null){ 46 try { 47 in.close(); 48 } catch (IOException e) { 49 e.printStackTrace(); 50 } 51 in = null; 52 } 53 if(out != null){ 54 out.close(); 55 out = null; 56 } 57 if(socket != null){ 58 try { 59 socket.close(); 60 } catch (IOException e) { 61 e.printStackTrace(); 62 } 63 socket = null; 64 } 65 } 66 } 67 }
客户端源码：
1 package com.anxpp.io.calculator.bio; 2 import java.io.BufferedReader; 3 import java.io.IOException; 4 import java.io.InputStreamReader; 5 import java.io.PrintWriter; 6 import java.net.Socket; 7 /** 8 * 阻塞式I/O创建的客户端 9 * @author yangtao__anxpp.com 10 * @version 1.0 11 */ 12 public class Client { 13 //默认的端口号 14 private static int DEFAULT_SERVER_PORT = 12345; 15 private static String DEFAULT_SERVER_IP = "127.0.0.1"; 16 public static void send(String expression){ 17 send(DEFAULT_SERVER_PORT,expression); 18 } 19 public static void send(int port,String expression){ 20 System.out.println("算术表达式为：" + expression); 21 Socket socket = null; 22 BufferedReader in = null; 23 PrintWriter out = null; 24 try{ 25 socket = new Socket(DEFAULT_SERVER_IP,port); 26 in = new BufferedReader(new InputStreamReader(socket.getInputStream())); 27 out = new PrintWriter(socket.getOutputStream(),true); 28 out.println(expression); 29 System.out.println("___结果为：" + in.readLine()); 30 }catch(Exception e){ 31 e.printStackTrace(); 32 }finally{ 33 //一下必要的清理工作 34 if(in != null){ 35 try { 36 in.close(); 37 } catch (IOException e) { 38 e.printStackTrace(); 39 } 40 in = null; 41 } 42 if(out != null){ 43 out.close(); 44 out = null; 45 } 46 if(socket != null){ 47 try { 48 socket.close(); 49 } catch (IOException e) { 50 e.printStackTrace(); 51 } 52 socket = null; 53 } 54 } 55 } 56 }
从以上代码，很容易看出，BIO主要的问题在于每当有一个新的客户端请求接入时，服务端必须创建一个新的线程来处理这条链路，在需要满足高性能、高并发的场景是没法应用的（大量创建新的线程会严重影响服务器性能，甚至罢工）。
1.2、伪异步I/O编程
为了改进这种一连接一线程的模型，我们可以使用线程池来管理这些线程，实现1个或多个线程处理N个客户端的模型（但是底层还是使用的同步阻塞I/O），通常被称为“伪异步I/O模型“。
伪异步I/O模型图：
实现很简单，我们只需要将新建线程的地方，交给线程池管理即可，只需要改动刚刚的Server代码即可：
1 package com.anxpp.io.calculator.bio; 2 import java.io.IOException; 3 import java.net.ServerSocket; 4 import java.net.Socket; 5 import java.util.concurrent.ExecutorService; 6 import java.util.concurrent.Executors; 7 /** 8 * BIO服务端源码__伪异步I/O 9 * @author yangtao__anxpp.com 10 * @version 1.0 11 */ 12 public final class ServerBetter { 13 //默认的端口号 14 private static int DEFAULT_PORT = 12345; 15 //单例的ServerSocket 16 private static ServerSocket server; 17 //线程池 懒汉式的单例 18 private static ExecutorService executorService = Executors.newFixedThreadPool(60); 19 //根据传入参数设置监听端口，如果没有参数调用以下方法并使用默认值 20 public static void start() throws IOException{ 21 //使用默认值 22 start(DEFAULT_PORT); 23 } 24 //这个方法不会被大量并发访问，不太需要考虑效率，直接进行方法同步就行了 25 public synchronized static void start(int port) throws IOException{ 26 if(server != null) return; 27 try{ 28 //通过构造函数创建ServerSocket 29 //如果端口合法且空闲，服务端就监听成功 30 server = new ServerSocket(port); 31 System.out.println("服务器已启动，端口号：" + port); 32 //通过无线循环监听客户端连接 33 //如果没有客户端接入，将阻塞在accept操作上。 34 while(true){ 35 Socket socket = server.accept(); 36 //当有新的客户端接入时，会执行下面的代码 37 //然后创建一个新的线程处理这条Socket链路 38 executorService.execute(new ServerHandler(socket)); 39 } 40 }finally{ 41 //一些必要的清理工作 42 if(server != null){ 43 System.out.println("服务器已关闭。"); 44 server.close(); 45 server = null; 46 } 47 } 48 } 49 }
我们知道，如果使用CachedThreadPool线程池（不限制线程数量，如果不清楚请参考文首提供的文章），其实除了能自动帮我们管理线程（复用），看起来也就像是1:1的客户端：线程数模型，而使用FixedThreadPool我们就有效的控制了线程的最大数量，保证了系统有限的资源的控制，实现了N:M的伪异步I/O模型。
但是，正因为限制了线程数量，如果发生大量并发请求，超过最大数量的线程就只能等待，直到线程池中的有空闲的线程可以被复用。而对Socket的输入流就行读取时，会一直阻塞，直到发生：
- 有数据可读,
- 可用数据以及读取完毕
- 发生空指针或I/O异常
所以在读取数据较慢时（比如数据量大、网络传输慢等），大量并发的情况下，其他接入的消息，只能一直等待，这就是最大的弊端。
2、NIO 编程 JDK 1.4中的java.nio.*包中引入新的Java I/O库，其目的是提高速度。实际上，“旧”的I/O包已经使用NIO重新实现过，即使我们不显式的使用NIO编程，也能从中受益。速度的提高在文件I/O和网络I/O中都可能会发生，但本文只讨论后者。
2.1、简介 NIO我们一般认为是New I/O（也是官方的叫法），因为它是相对于老的I/O类库新增的（其实在JDK 1.4中就已经被引入了，但这个名词还会继续用很久，即使它们在现在看来已经是“旧”的了，所以也提示我们在命名时，需要好好考虑），做了很大的改变。但民间跟多人称之为Non-block I/O，即非阻塞I/O，因为这样叫，更能体现它的特点。而下文中的NIO，不是指整个新的I/O库，而是非阻塞I/O。
NIO提供了与传统BIO模型中的Socket和ServerSocket相对应的SocketChannel和ServerSocketChannel两种不同的套接字通道实现。
新增的着两种通道都支持阻塞和非阻塞两种模式。
阻塞模式使用就像传统中的支持一样，比较简单，但是性能和可靠性都不好；非阻塞模式正好与之相反。
对于低负载、低并发的应用程序，可以使用同步阻塞I/O来提升开发速率和更好的维护性；对于高负载、高并发的（网络）应用，应使用NIO的非阻塞模式来开发。
下面会先对基础知识进行介绍。
2.2、缓冲区 Buffer Buffer是一个对象，包含一些要写入或者读出的数据。
在NIO库中，所有数据都是用缓冲区处理的。在读取数据时，它是直接读到缓冲区中的；在写入数据时，也是写入到缓冲区中。任何时候访问NIO中的数据，都是通过缓冲区进行操作。
缓冲区实际上是一个数组，并提供了对数据结构化访问以及维护读写位置等信息。
具体的缓存区有这些：ByteBuffe、CharBuffer、 ShortBuffer、IntBuffer、LongBuffer、FloatBuffer、DoubleBuffer。他们实现了相同的接口：Buffer。
2.3、通道 Channel
我们对数据的读取和写入要通过Channel，它就像水管一样，是一个通道。通道不同于流的地方就是通道是双向的，可以用于读、写和同时读写操作。
底层的操作系统的通道一般都是全双工的，所以全双工的Channel比流能更好的映射底层操作系统的API。
Channel主要分两大类： SelectableChannel：用户网络读写 FileChannel：用于文件操作 , 后面代码会涉及的ServerSocketChannel和SocketChannel都是SelectableChannel的子类。
2.4、多路复用器 Selector
Selector是Java NIO 编程的基础。
Selector提供选择已经就绪的任务的能力：Selector会不断轮询注册在其上的Channel，如果某个Channel上面发生读或者写事件，这个Channel就处于就绪状态，会被Selector轮询出来，然后通过SelectionKey可以获取就绪Channel的集合，进行后续的I/O操作。
一个Selector可以同时轮询多个Channel，因为JDK使用了epoll()代替传统的select实现，所以没有最大连接句柄1024/2048的限制。所以，只需要一个线程负责Selector的轮询，就可以接入成千上万的客户端。
2.5、NIO服务端
代码比传统的Socket编程看起来要复杂不少。
直接贴代码吧，以注释的形式给出代码说明。
NIO创建的Server源码：
1 package com.anxpp.io.calculator.nio; 2 public class Server { 3 private static int DEFAULT_PORT = 12345; 4 private static ServerHandle serverHandle; 5 public static void start(){ 6 start(DEFAULT_PORT); 7 } 8 public static synchronized void start(int port){ 9 if(serverHandle!=null) 10 serverHandle.stop(); 11 serverHandle = new ServerHandle(port); 12 new Thread(serverHandle,"Server").start(); 13 } 14 public static void main(String[] args){ 15 start(); 16 } 17 }
1 package com.anxpp.io.calculator.nio; 2 import java.io.IOException; 3 import java.net.InetSocketAddress; 4 import java.nio.ByteBuffer; 5 import java.nio.channels.SelectionKey; 6 import java.nio.channels.Selector; 7 import java.nio.channels.ServerSocketChannel; 8 import java.nio.channels.SocketChannel; 9 import java.util.Iterator; 10 import java.util.Set; 11 12 import com.anxpp.io.utils.Calculator; 13 /** 14 * NIO服务端 15 * @author yangtao__anxpp.com 16 * @version 1.0 17 */ 18 public class ServerHandle implements Runnable{ 19 private Selector selector; 20 private ServerSocketChannel serverChannel; 21 private volatile boolean started; 22 /** 23 * 构造方法 24 * @param port 指定要监听的端口号 25 */ 26 public ServerHandle(int port) { 27 try{ 28 //创建选择器 29 selector = Selector.open(); 30 //打开监听通道 31 serverChannel = ServerSocketChannel.open(); 32 //如果为 true，则此通道将被置于阻塞模式；如果为 false，则此通道将被置于非阻塞模式 33 serverChannel.configureBlocking(false);//开启非阻塞模式 34 //绑定端口 backlog设为1024 35 serverChannel.socket().bind(new InetSocketAddress(port),1024); 36 //监听客户端连接请求 37 serverChannel.register(selector, SelectionKey.OP_ACCEPT); 38 //标记服务器已开启 39 started = true; 40 System.out.println("服务器已启动，端口号：" + port); 41 }catch(IOException e){ 42 e.printStackTrace(); 43 System.exit(1); 44 } 45 } 46 public void stop(){ 47 started = false; 48 } 49 @Override 50 public void run() { 51 //循环遍历selector 52 while(started){ 53 try{ 54 //无论是否有读写事件发生，selector每隔1s被唤醒一次 55 selector.select(1000); 56 //阻塞,只有当至少一个注册的事件发生的时候才会继续. 57 // selector.select(); 58 Set<SelectionKey> keys = selector.selectedKeys(); 59 Iterator<SelectionKey> it = keys.iterator(); 60 SelectionKey key = null; 61 while(it.hasNext()){ 62 key = it.next(); 63 it.remove(); 64 try{ 65 handleInput(key); 66 }catch(Exception e){ 67 if(key != null){ 68 key.cancel(); 69 if(key.channel() != null){ 70 key.channel().close(); 71 } 72 } 73 } 74 } 75 }catch(Throwable t){ 76 t.printStackTrace(); 77 } 78 } 79 //selector关闭后会自动释放里面管理的资源 80 if(selector != null) 81 try{ 82 selector.close(); 83 }catch (Exception e) { 84 e.printStackTrace(); 85 } 86 } 87 private void handleInput(SelectionKey key) throws IOException{ 88 if(key.isValid()){ 89 //处理新接入的请求消息 90 if(key.isAcceptable()){ 91 ServerSocketChannel ssc = (ServerSocketChannel) key.channel(); 92 //通过ServerSocketChannel的accept创建SocketChannel实例 93 //完成该操作意味着完成TCP三次握手，TCP物理链路正式建立 94 SocketChannel sc = ssc.accept(); 95 //设置为非阻塞的 96 sc.configureBlocking(false); 97 //注册为读 98 sc.register(selector, SelectionKey.OP_READ); 99 } 100 //读消息 101 if(key.isReadable()){ 102 SocketChannel sc = (SocketChannel) key.channel(); 103 //创建ByteBuffer，并开辟一个1M的缓冲区 104 ByteBuffer buffer = ByteBuffer.allocate(1024); 105 //读取请求码流，返回读取到的字节数 106 int readBytes = sc.read(buffer); 107 //读取到字节，对字节进行编解码 108 if(readBytes>0){ 109 //将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作 110 buffer.flip(); 111 //根据缓冲区可读字节数创建字节数组 112 byte[] bytes = new byte[buffer.remaining()]; 113 //将缓冲区可读字节数组复制到新建的数组中 114 buffer.get(bytes); 115 String expression = new String(bytes,"UTF-8"); 116 System.out.println("服务器收到消息：" + expression); 117 //处理数据 118 String result = null; 119 try{ 120 result = Calculator.cal(expression).toString(); 121 }catch(Exception e){ 122 result = "计算错误：" + e.getMessage(); 123 } 124 //发送应答消息 125 doWrite(sc,result); 126 } 127 //没有读取到字节 忽略 128 // else if(readBytes==0); 129 //链路已经关闭，释放资源 130 else if(readBytes<0){ 131 key.cancel(); 132 sc.close(); 133 } 134 } 135 } 136 } 137 //异步发送应答消息 138 private void doWrite(SocketChannel channel,String response) throws IOException{ 139 //将消息编码为字节数组 140 byte[] bytes = response.getBytes(); 141 //根据数组容量创建ByteBuffer 142 ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length); 143 //将字节数组复制到缓冲区 144 writeBuffer.put(bytes); 145 //flip操作 146 writeBuffer.flip(); 147 //发送缓冲区的字节数组 148 channel.write(writeBuffer); 149 //****此处不含处理“写半包”的代码 150 } 151 }
1 package com.anxpp.io.calculator.nio; 2 public class Client { 3 private static String DEFAULT_HOST = "127.0.0.1"; 4 private static int DEFAULT_PORT = 12345; 5 private static ClientHandle clientHandle; 6 public static void start(){ 7 start(DEFAULT_HOST,DEFAULT_PORT); 8 } 9 public static synchronized void start(String ip,int port){ 10 if(clientHandle!=null) 11 clientHandle.stop(); 12 clientHandle = new ClientHandle(ip,port); 13 new Thread(clientHandle,"Server").start(); 14 } 15 //向服务器发送消息 16 public static boolean sendMsg(String msg) throws Exception{ 17 if(msg.equals("q")) return false; 18 clientHandle.sendMsg(msg); 19 return true; 20 } 21 public static void main(String[] args){ 22 start(); 23 } 24 }
1 package com.anxpp.io.calculator.nio; 2 import java.io.IOException; 3 import java.net.InetSocketAddress; 4 import java.nio.ByteBuffer; 5 import java.nio.channels.SelectionKey; 6 import java.nio.channels.Selector; 7 import java.nio.channels.SocketChannel; 8 import java.util.Iterator; 9 import java.util.Set; 10 /** 11 * NIO客户端 12 * @author yangtao__anxpp.com 13 * @version 1.0 14 */ 15 public class ClientHandle implements Runnable{ 16 private String host; 17 private int port; 18 private Selector selector; 19 private SocketChannel socketChannel; 20 private volatile boolean started; 21 22 public ClientHandle(String ip,int port) { 23 this.host = ip; 24 this.port = port; 25 try{ 26 //创建选择器 27 selector = Selector.open(); 28 //打开监听通道 29 socketChannel = SocketChannel.open(); 30 //如果为 true，则此通道将被置于阻塞模式；如果为 false，则此通道将被置于非阻塞模式 31 socketChannel.configureBlocking(false);//开启非阻塞模式 32 started = true; 33 }catch(IOException e){ 34 e.printStackTrace(); 35 System.exit(1); 36 } 37 } 38 public void stop(){ 39 started = false; 40 } 41 @Override 42 public void run() { 43 try{ 44 doConnect(); 45 }catch(IOException e){ 46 e.printStackTrace(); 47 System.exit(1); 48 } 49 //循环遍历selector 50 while(started){ 51 try{ 52 //无论是否有读写事件发生，selector每隔1s被唤醒一次 53 selector.select(1000); 54 //阻塞,只有当至少一个注册的事件发生的时候才会继续. 55 // selector.select(); 56 Set<SelectionKey> keys = selector.selectedKeys(); 57 Iterator<SelectionKey> it = keys.iterator(); 58 SelectionKey key = null; 59 while(it.hasNext()){ 60 key = it.next(); 61 it.remove(); 62 try{ 63 handleInput(key); 64 }catch(Exception e){ 65 if(key != null){ 66 key.cancel(); 67 if(key.channel() != null){ 68 key.channel().close(); 69 } 70 } 71 } 72 } 73 }catch(Exception e){ 74 e.printStackTrace(); 75 System.exit(1); 76 } 77 } 78 //selector关闭后会自动释放里面管理的资源 79 if(selector != null) 80 try{ 81 selector.close(); 82 }catch (Exception e) { 83 e.printStackTrace(); 84 } 85 } 86 private void handleInput(SelectionKey key) throws IOException{ 87 if(key.isValid()){ 88 SocketChannel sc = (SocketChannel) key.channel(); 89 if(key.isConnectable()){ 90 if(sc.finishConnect()); 91 else System.exit(1); 92 } 93 //读消息 94 if(key.isReadable()){ 95 //创建ByteBuffer，并开辟一个1M的缓冲区 96 ByteBuffer buffer = ByteBuffer.allocate(1024); 97 //读取请求码流，返回读取到的字节数 98 int readBytes = sc.read(buffer); 99 //读取到字节，对字节进行编解码 100 if(readBytes>0){ 101 //将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作 102 buffer.flip(); 103 //根据缓冲区可读字节数创建字节数组 104 byte[] bytes = new byte[buffer.remaining()]; 105 //将缓冲区可读字节数组复制到新建的数组中 106 buffer.get(bytes); 107 String result = new String(bytes,"UTF-8"); 108 System.out.println("客户端收到消息：" + result); 109 } 110 //没有读取到字节 忽略 111 // else if(readBytes==0); 112 //链路已经关闭，释放资源 113 else if(readBytes<0){ 114 key.cancel(); 115 sc.close(); 116 } 117 } 118 } 119 } 120 //异步发送消息 121 private void doWrite(SocketChannel channel,String request) throws IOException{ 122 //将消息编码为字节数组 123 byte[] bytes = request.getBytes(); 124 //根据数组容量创建ByteBuffer 125 ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length); 126 //将字节数组复制到缓冲区 127 writeBuffer.put(bytes); 128 //flip操作 129 writeBuffer.flip(); 130 //发送缓冲区的字节数组 131 channel.write(writeBuffer); 132 //****此处不含处理“写半包”的代码 133 } 134 private void doConnect() throws IOException{ 135 if(socketChannel.connect(new InetSocketAddress(host,port))); 136 else socketChannel.register(selector, SelectionKey.OP_CONNECT); 137 } 138 public void sendMsg(String msg) throws Exception{ 139 socketChannel.register(selector, SelectionKey.OP_READ); 140 doWrite(socketChannel, msg); 141 } 142 }
可以看到，创建NIO服务端的主要步骤如下：
- 打开ServerSocketChannel，监听客户端连接
- 绑定监听端口，设置连接为非阻塞模式
- 创建Reactor线程，创建多路复用器并启动线程
- 将ServerSocketChannel注册到Reactor线程中的Selector上，监听ACCEPT事件
- Selector轮询准备就绪的key
- Selector监听到新的客户端接入，处理新的接入请求，完成TCP三次握手，简历物理链路
- 设置客户端链路为非阻塞模式
- 将新接入的客户端连接注册到Reactor线程的Selector上，监听读操作，读取客户端发送的网络消息
- 异步读取客户端消息到缓冲区
- 对Buffer编解码，处理半包消息，将解码成功的消息封装成Task
- 将应答消息编码为Buffer，调用SocketChannel的write将消息异步发送给客户端
因为应答消息的发送，SocketChannel也是异步非阻塞的，所以不能保证一次能吧需要发送的数据发送完，此时就会出现写半包的问题。我们需要注册写操作，不断轮询Selector将没有发送完的消息发送完毕，然后通过Buffer的hasRemain()方法判断消息是否发送完成。
3、AIO编程 NIO 2.0引入了新的异步通道的概念，并提供了异步文件通道和异步套接字通道的实现。
异步的套接字通道时真正的异步非阻塞I/O，对应于UNIX网络编程中的事件驱动I/O（AIO）。他不需要过多的Selector对注册的通道进行轮询即可实现异步读写，从而简化了NIO的编程模型。
1 package com.anxpp.io.calculator.aio.server; 2 /** 3 * AIO服务端 4 * @author yangtao__anxpp.com 5 * @version 1.0 6 */ 7 public class Server { 8 private static int DEFAULT_PORT = 12345; 9 private static AsyncServerHandler serverHandle; 10 public volatile static long clientCount = 0; 11 public static void start(){ 12 start(DEFAULT_PORT); 13 } 14 public static synchronized void start(int port){ 15 if(serverHandle!=null) 16 return; 17 serverHandle = new AsyncServerHandler(port); 18 new Thread(serverHandle,"Server").start(); 19 } 20 public static void main(String[] args){ 21 Server.start(); 22 } 23 }
1 package com.anxpp.io.calculator.aio.server; 2 import java.io.IOException; 3 import java.net.InetSocketAddress; 4 import java.nio.channels.AsynchronousServerSocketChannel; 5 import java.util.concurrent.CountDownLatch; 6 public class AsyncServerHandler implements Runnable { 7 public CountDownLatch latch; 8 public AsynchronousServerSocketChannel channel; 9 public AsyncServerHandler(int port) { 10 try { 11 //创建服务端通道 12 channel = AsynchronousServerSocketChannel.open(); 13 //绑定端口 14 channel.bind(new InetSocketAddress(port)); 15 System.out.println("服务器已启动，端口号：" + port); 16 } catch (IOException e) { 17 e.printStackTrace(); 18 } 19 } 20 @Override 21 public void run() { 22 //CountDownLatch初始化 23 //它的作用：在完成一组正在执行的操作之前，允许当前的现场一直阻塞 24 //此处，让现场在此阻塞，防止服务端执行完成后退出 25 //也可以使用while(true)+sleep 26 //生成环境就不需要担心这个问题，以为服务端是不会退出的 27 latch = new CountDownLatch(1); 28 //用于接收客户端的连接 29 channel.accept(this,new AcceptHandler()); 30 try { 31 latch.await(); 32 } catch (InterruptedException e) { 33 e.printStackTrace(); 34 } 35 } 36 }
1 package com.anxpp.io.calculator.aio.server; 2 import java.nio.ByteBuffer; 3 import java.nio.channels.AsynchronousSocketChannel; 4 import java.nio.channels.CompletionHandler; 5 //作为handler接收客户端连接 6 public class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncServerHandler> { 7 @Override 8 public void completed(AsynchronousSocketChannel channel,AsyncServerHandler serverHandler) { 9 //继续接受其他客户端的请求 10 Server.clientCount++; 11 System.out.println("连接的客户端数：" + Server.clientCount); 12 serverHandler.channel.accept(serverHandler, this); 13 //创建新的Buffer 14 ByteBuffer buffer = ByteBuffer.allocate(1024); 15 //异步读 第三个参数为接收消息回调的业务Handler 16 channel.read(buffer, buffer, new ReadHandler(channel)); 17 } 18 @Override 19 public void failed(Throwable exc, AsyncServerHandler serverHandler) { 20 exc.printStackTrace(); 21 serverHandler.latch.countDown(); 22 } 23 }
1 package com.anxpp.io.calculator.aio.server; 2 import java.io.IOException; 3 import java.io.UnsupportedEncodingException; 4 import java.nio.ByteBuffer; 5 import java.nio.channels.AsynchronousSocketChannel; 6 import java.nio.channels.CompletionHandler; 7 import com.anxpp.io.utils.Calculator; 8 public class ReadHandler implements CompletionHandler<Integer, ByteBuffer> { 9 //用于读取半包消息和发送应答 10 private AsynchronousSocketChannel channel; 11 public ReadHandler(AsynchronousSocketChannel channel) { 12 this.channel = channel; 13 } 14 //读取到消息后的处理 15 @Override 16 public void completed(Integer result, ByteBuffer attachment) { 17 //flip操作 18 attachment.flip(); 19 //根据 20 byte[] message = new byte[attachment.remaining()]; 21 attachment.get(message); 22 try { 23 String expression = new String(message, "UTF-8"); 24 System.out.println("服务器收到消息: " + expression); 25 String calrResult = null; 26 try{ 27 calrResult = Calculator.cal(expression).toString(); 28 }catch(Exception e){ 29 calrResult = "计算错误：" + e.getMessage(); 30 } 31 //向客户端发送消息 32 doWrite(calrResult); 33 } catch (UnsupportedEncodingException e) { 34 e.printStackTrace(); 35 } 36 } 37 //发送消息 38 private void doWrite(String result) { 39 byte[] bytes = result.getBytes(); 40 ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length); 41 writeBuffer.put(bytes); 42 writeBuffer.flip(); 43 //异步写数据 参数与前面的read一样 44 channel.write(writeBuffer, writeBuffer,new CompletionHandler<Integer, ByteBuffer>() { 45 @Override 46 public void completed(Integer result, ByteBuffer buffer) { 47 //如果没有发送完，就继续发送直到完成 48 if (buffer.hasRemaining()) 49 channel.write(buffer, buffer, this); 50 else{ 51 //创建新的Buffer 52 ByteBuffer readBuffer = ByteBuffer.allocate(1024); 53 //异步读 第三个参数为接收消息回调的业务Handler 54 channel.read(readBuffer, readBuffer, new ReadHandler(channel)); 55 } 56 } 57 @Override 58 public void failed(Throwable exc, ByteBuffer attachment) { 59 try { 60 channel.close(); 61 } catch (IOException e) { 62 } 63 } 64 }); 65 } 66 @Override 67 public void failed(Throwable exc, ByteBuffer attachment) { 68 try { 69 this.channel.close(); 70 } catch (IOException e) { 71 e.printStackTrace(); 72 } 73 } 74 }
1 package com.anxpp.io.calculator.aio.client; 2 import java.util.Scanner; 3 public class Client { 4 private static String DEFAULT_HOST = "127.0.0.1"; 5 private static int DEFAULT_PORT = 12345; 6 private static AsyncClientHandler clientHandle; 7 public static void start(){ 8 start(DEFAULT_HOST,DEFAULT_PORT); 9 } 10 public static synchronized void start(String ip,int port){ 11 if(clientHandle!=null) 12 return; 13 clientHandle = new AsyncClientHandler(ip,port); 14 new Thread(clientHandle,"Client").start(); 15 } 16 //向服务器发送消息 17 public static boolean sendMsg(String msg) throws Exception{ 18 if(msg.equals("q")) return false; 19 clientHandle.sendMsg(msg); 20 return true; 21 } 22 @SuppressWarnings("resource") 23 public static void main(String[] args) throws Exception{ 24 Client.start(); 25 System.out.println("请输入请求消息："); 26 Scanner scanner = new Scanner(System.in); 27 while(Client.sendMsg(scanner.nextLine())); 28 } 29 }
1 package com.anxpp.io.calculator.aio.client; 2 import java.io.IOException; 3 import java.net.InetSocketAddress; 4 import java.nio.ByteBuffer; 5 import java.nio.channels.AsynchronousSocketChannel; 6 import java.nio.channels.CompletionHandler; 7 import java.util.concurrent.CountDownLatch; 8 public class AsyncClientHandler implements CompletionHandler<Void, AsyncClientHandler>, Runnable { 9 private AsynchronousSocketChannel clientChannel; 10 private String host; 11 private int port; 12 private CountDownLatch latch; 13 public AsyncClientHandler(String host, int port) { 14 this.host = host; 15 this.port = port; 16 try { 17 //创建异步的客户端通道 18 clientChannel = AsynchronousSocketChannel.open(); 19 } catch (IOException e) { 20 e.printStackTrace(); 21 } 22 } 23 @Override 24 public void run() { 25 //创建CountDownLatch等待 26 latch = new CountDownLatch(1); 27 //发起异步连接操作，回调参数就是这个类本身，如果连接成功会回调completed方法 28 clientChannel.connect(new InetSocketAddress(host, port), this, this); 29 try { 30 latch.await(); 31 } catch (InterruptedException e1) { 32 e1.printStackTrace(); 33 } 34 try { 35 clientChannel.close(); 36 } catch (IOException e) { 37 e.printStackTrace(); 38 } 39 } 40 //连接服务器成功 41 //意味着TCP三次握手完成 42 @Override 43 public void completed(Void result, AsyncClientHandler attachment) { 44 System.out.println("客户端成功连接到服务器..."); 45 } 46 //连接服务器失败 47 @Override 48 public void failed(Throwable exc, AsyncClientHandler attachment) { 49 System.err.println("连接服务器失败..."); 50 exc.printStackTrace(); 51 try { 52 clientChannel.close(); 53 latch.countDown(); 54 } catch (IOException e) { 55 e.printStackTrace(); 56 } 57 } 58 //向服务器发送消息 59 public void sendMsg(String msg){ 60 byte[] req = msg.getBytes(); 61 ByteBuffer writeBuffer = ByteBuffer.allocate(req.length); 62 writeBuffer.put(req); 63 writeBuffer.flip(); 64 //异步写 65 clientChannel.write(writeBuffer, writeBuffer,new WriteHandler(clientChannel, latch)); 66 } 67 }
package com.anxpp.io.calculator.aio.client; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousSocketChannel; import java.nio.channels.CompletionHandler; import java.util.concurrent.CountDownLatch; public class WriteHandler implements CompletionHandler<Integer, ByteBuffer> { private AsynchronousSocketChannel clientChannel; private CountDownLatch latch; public WriteHandler(AsynchronousSocketChannel clientChannel,CountDownLatch latch) { this.clientChannel = clientChannel; this.latch = latch; } @Override public void completed(Integer result, ByteBuffer buffer) { //完成全部数据的写入 if (buffer.hasRemaining()) { clientChannel.write(buffer, buffer, this); } else { //读取数据 ByteBuffer readBuffer = ByteBuffer.allocate(1024); clientChannel.read(readBuffer,readBuffer,new ReadHandler(clientChannel, latch)); } } @Override public void failed(Throwable exc, ByteBuffer attachment) { System.err.println("数据发送失败..."); try { clientChannel.close(); latch.countDown(); } catch (IOException e) { } } }
1 package com.anxpp.io.calculator.aio.client; 2 import java.io.IOException; 3 import java.io.UnsupportedEncodingException; 4 import java.nio.ByteBuffer; 5 import java.nio.channels.AsynchronousSocketChannel; 6 import java.nio.channels.CompletionHandler; 7 import java.util.concurrent.CountDownLatch; 8 public class ReadHandler implements CompletionHandler<Integer, ByteBuffer> { 9 private AsynchronousSocketChannel clientChannel; 10 private CountDownLatch latch; 11 public ReadHandler(AsynchronousSocketChannel clientChannel,CountDownLatch latch) { 12 this.clientChannel = clientChannel; 13 this.latch = latch; 14 } 15 @Override 16 public void completed(Integer result,ByteBuffer buffer) { 17 buffer.flip(); 18 byte[] bytes = new byte[buffer.remaining()]; 19 buffer.get(bytes); 20 String body; 21 try { 22 body = new String(bytes,"UTF-8"); 23 System.out.println("客户端收到结果:"+ body); 24 } catch (UnsupportedEncodingException e) { 25 e.printStackTrace(); 26 } 27 } 28 @Override 29 public void failed(Throwable exc,ByteBuffer attachment) { 30 System.err.println("数据读取失败..."); 31 try { 32 clientChannel.close(); 33 latch.countDown(); 34 } catch (IOException e) { 35 } 36 } 37 }
4、各种I/O的对比 先以一张表来直观的对比一下：
具体选择什么样的模型或者NIO框架，完全基于业务的实际应用场景和性能需求，如果客户端很少，服务器负荷不重，就没有必要选择开发起来相对不那么简单的NIO做服务端；相反，就应考虑使用NIO或者相关的框架了。
--------------------- 转自：CSDN 原文：https://blog.csdn.net/anxpp/article/details/51512200
https://www.cnblogs.com/xiaoxi/p/6525396.html
以上是关于BIO,NIO区别的主要内容，如果未能解决你的问题，请参考以下文章