本帖最后由 长沙-小知姐姐 于 2019-2-21 15:00 编辑
1. I/O模型分类：
下图所示各种IO差异：
下面我们聊聊,阻塞和非阻塞，同步和异步。
2. 什么是同步和异步：
同步和异步是针对应用程序和内核的交互而言的，同步指的是用户进程触发IO操作并等待或者轮询的去查看IO操作是否就绪，而异步是指用户进程触发IO操作以后便开始做自己的事情，而当IO操作已经完成的时候会得到IO完成的通知。
3. 什么是阻塞和非阻塞：
阻塞和非阻塞是针对于进程在访问数据的时候，根据IO操作的就绪状态来采取的不同方式，说白了是一种读取或者写入操作函数的实现方式，阻塞方式下读取或者写入函数将一直等待，而非阻塞方式下，读取或者写入函数会立即返回一个状态值。
4. 同步阻塞IO（BIO）：
在此种方式下，用户进程在发起一个IO操作以后，必须等待IO操作的完成，只有当真正完成了IO操作以后，用户进程才能运行。我们学习的JAVA传统的IO模型属于此种方式！网络编程的基本模型是C/S模型，即两个进程间的通信。服务端提供IP和监听端口，客户端通过连接操作想服务端监听的地址发起连接请求，通过三次握手连接，如果连接成功建立，双方就可以通过套接字进行通信。传统的同步阻塞模型开发中，ServerSocket负责绑定IP地址，启动监听端口；Socket负责发起连接操作。连接成功后，双方通过输入和输出流进行同步阻塞式通信。简单的描述一下BIO的服务端通信模型：采用BIO通信模型的服务端，通常由一个独立的Acceptor线程负责监听客户端的连接，它接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理没处理完成后，通过输出流返回应答给客户端，线程销毁。即典型的一请求一应答的模型。
代码
```java
Server端： package com.itcast.demo1;
import java.io.*; import java.net.ServerSocket; import java.net.Socket;
public class Server { public static void main(String[] args) throws IOException { ServerSocket ss = new ServerSocket(10086); while(true){ //监听，等待客户端连接 Socket s = ss.accept(); new Thread(()->{ try { //读取客户端发送过来的数据 InputStream is = s.getInputStream(); //接受客户端传过来io流数据 //本地io流，把客户端上传过来的文件，保存到我服务端的本地 //F:\\Upload File f = new File("f:\\upload"); if (!f.exists()) f.mkdirs(); long l = System.currentTimeMillis(); BufferedOutputStream bos = new BufferedOutputStream( new FileOutputStream(new File(f, "c"+l+".gif"))); byte[] arr = new byte[1024 * 8]; int len ;//arr = {-1}' 1 while((len = is.read(arr))!= -1){ bos.write(arr,0,len); } //写回一个数据给客户端 OutputStream os = s.getOutputStream(); PrintStream ps = new PrintStream(os, true); ps.println("上传成功~"); s.close(); } catch (IOException e) { e.printStackTrace(); } }).start(); } } }
```
```java
Client端： package com.itcast.demo1; import java.io.*; import java.net.Socket;
public class Client { public static void main(String[] args) throws IOException { //创建Socket对象 Socket sc = new Socket("localhost",10086); //向服务器写数据 // 用IO从本地读取一个图片 FileInputStream fis = new FileInputStream("f:\\a.gif"); OutputStream os = sc.getOutputStream(); byte[] arr = new byte[1024*4]; int len ; //一边读，一边写到服务器 while((len = fis.read(arr))!= -1){ os.write(arr,0,len); } sc.shutdownOutput();//相当于告诉服务端，我客户端这边已经写完了 //读取服务器返回的数据 InputStream is = sc.getInputStream(); BufferedReader br = new BufferedReader(new InputStreamReader(is)); String s = br.readLine(); System.out.println(s); fis.close(); sc.close(); } }
```
简单代码，看似没问题，但是该模型最大的问题就是缺乏弹性伸缩能力，当客户端并发访问量增加后，服务端的线程个数和客户端并发访问数呈1:1的正比关系，Java中的线程也是比较宝贵的系统资源，线程数量快速膨胀后，系统的性能将急剧下降，随着访问量的继续增大，系统最终就挂掉。
5. 同步非阻塞IO（NIO）
在此种方式下，用户进程发起一个IO操作以后边可返回做其它事情，但是用户进程需要时不时的询问IO操作是否就绪，这就要求用户进程不停的去询问，从而引入不必要的CPU资源浪费。其中目前JAVA的NIO就属于同步非阻塞IO。 NIO官方的叫法New I/O，它是相对于老的I/O类库新增的，做了很大的改变。但民间跟多人称之为Non-block I/O，即非阻塞I/O，因为这样叫，更能体现它的特点。而下文中的NIO，不是指整个新的I/O库，而是非阻塞I/O。NIO提供了与传统BIO模型中相对应的套接字： Socket ->> SocketChannel ServerSocke ->> ServerSocketChannel 并且新增的着两种通道都支持阻塞和非阻塞两种模式。 阻塞模式使用就像传统中的支持一样，比较简单，但是性能和可靠性都不好；非阻塞模式正好与之相反。对于低负载、低并发的应用程序，可以使用同步阻塞I/O来提升开发速率和更好的维护性；对于高负载、高并发的（网络）应用，应使用NIO的非阻塞模式来开发。下面会先对基础知识进行介绍。
5.1 缓冲区 BufferBuffer是一个对象，包含一些要写入或者读出的数据。 在NIO库中，所有数据都是用缓冲区处理的。在读取数据时，它是直接读到缓冲区中的；在写入数据时，也是写入到缓冲区中。任何时候访问NIO中的数据，都是通过缓冲区进行操作。 缓冲区实际上是一个数组，并提供了对数据结构化访问以及维护读写位置等信息。 具体的缓存区有这些：ByteBuffe、CharBuffer、 ShortBuffer、IntBuffer、LongBuffer、FloatBuffer、DoubleBuffer。他们实现了相同的接口：Buffer。
5.2 通道 Channel我们对数据的读取和写入要通过Channel，它就像水管一样，是一个通道。通道不同于流的地方就是通道是双向的，可以用于读、写和同时读写操作。 底层的操作系统的通道一般都是全双工的，所以全双工的Channel比流能更好的映射底层操作系统的API。 Channel主要分两大类：SelectableChannel：用户网络读写 FileChannel：用于文件操作 后面代码会涉及的ServerSocketChannel和SocketChannel都是SelectableChannel的子类。
5.3 多路复用器 SelectorSelector是Java NIO 编程的基础。 Selector提供选择已经就绪的任务的能力：Selector会不断轮询注册在其上的Channel，如果某个Channel上面发生读或者写事件，这个Channel就处于就绪状态，会被Selector轮询出来，然后通过SelectionKey可以获取就绪Channel的集合，进行后续的I/O操作。 一个Selector可以同时轮询多个Channel，因为JDK使用了epoll()代替传统的select实现，所以没有最大连接句柄1024/2048的限制。所以，只需要一个线程负责Selector的轮询，就可以接入成千上万的客户端。 代码实例： 依旧分为服务端、客户端以及测试类：
Server端： package cn.itcast.nio;
public class Server { private static int DEFAULT_PORT = 12345; private static ServerHandle serverHandle; public static void start(){ start(DEFAULT_PORT); } public static synchronized void start(int port){ if(serverHandle!=null) serverHandle.stop(); serverHandle = new ServerHandle(port); Thread thread = new Thread(serverHandle); System.out.println(thread.getName()); thread.start(); } public static void main(String[] args){ start(); } }
package cn.itcast.nio;
import cn.itcast.util.Calculator;
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 ServerHandle implements Runnable{ //多路复用器 private Selector selector; //通道 private ServerSocketChannel serverChannel; private volatile boolean started; /** * 构造方法 * @param port 指定要监听的端口号 */ public ServerHandle(int port) { try{ //创建选择器 selector = Selector.open(); //打开监听通道 serverChannel = ServerSocketChannel.open(); //如果为 true，则此通道将被置于阻塞模式；如果为 false，则此通道将被置于非阻塞模式 serverChannel.configureBlocking(false);//开启非阻塞模式 //绑定端口 backlog设为1024 serverChannel.socket().bind(new InetSocketAddress(port),1024); //监听客户端连接请求 serverChannel.register(selector, SelectionKey.OP_ACCEPT); //标记服务器已开启 started = true; System.out.println("服务器已启动，端口号：" + port); }catch(IOException e){ e.printStackTrace(); System.exit(1); } }
public void stop(){ started = false; }
@Override public void run() { //循环遍历selector while(started){ try{ //无论是否有读写事件发生，selector每隔1s被唤醒一次 selector.select(1000); //阻塞,只有当至少一个注册的事件发生的时候才会继续. // selector.select(); Set keys = selector.selectedKeys(); Iterator it = keys.iterator(); SelectionKey key = null; while(it.hasNext()){ key =(SelectionKey) it.next(); it.remove(); try{ handleInput(key); }catch(Exception e){ if(key != null){ key.cancel(); if(key.channel() != null){ key.channel().close(); } } } } }catch(Throwable t){ t.printStackTrace(); } } //selector关闭后会自动释放里面管理的资源 if(selector != null) try{ selector.close(); }catch (Exception e) { e.printStackTrace(); } } private void handleInput(SelectionKey key) throws IOException{ if(key.isValid()){ //处理新接入的请求消息 if(key.isAcceptable()){ ServerSocketChannel ssc = (ServerSocketChannel) key.channel(); //通过ServerSocketChannel的accept创建SocketChannel实例 //完成该操作意味着完成TCP三次握手，TCP物理链路正式建立 SocketChannel sc = ssc.accept(); //设置为非阻塞的 sc.configureBlocking(false); //注册为读 sc.register(selector, SelectionKey.OP_READ); } //读消息 if(key.isReadable()){ SocketChannel sc = (SocketChannel) key.channel(); //创建ByteBuffer，并开辟一个1M的缓冲区 ByteBuffer buffer = ByteBuffer.allocate(1024); //读取请求码流，返回读取到的字节数 int readBytes = sc.read(buffer); //读取到字节，对字节进行编解码 if(readBytes>0){ //将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作 buffer.flip(); //根据缓冲区可读字节数创建字节数组 byte[] bytes = new byte[buffer.remaining()]; //将缓冲区可读字节数组复制到新建的数组中 buffer.get(bytes); String expression = new String(bytes,"UTF-8"); System.out.println("服务器收到消息：" + expression); //处理数据 String result = null; try{ result = Calculator.cal(expression).toString(); }catch(Exception e){ result = "计算错误：" + e.getMessage(); } //发送应答消息 doWrite(sc,result); } //链路已经关闭，释放资源 else if(readBytes<0){ key.cancel(); sc.close(); } } } } //异步发送应答消息 private void doWrite(SocketChannel channel,String response) throws IOException{ //将消息编码为字节数组 byte[] bytes = response.getBytes(); //根据数组容量创建ByteBuffer ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length); //将字节数组复制到缓冲区 writeBuffer.put(bytes); //flip操作 writeBuffer.flip(); //发送缓冲区的字节数组 channel.write(writeBuffer); } } ```
```java
Client端： package cn.itcast.nio;
public class Client { private static String DEFAULT_HOST = "127.0.0.1"; private static int DEFAULT_PORT = 12345; private static ClientHandle clientHandle; public static void start(){ start(DEFAULT_HOST,DEFAULT_PORT); } public static synchronized void start(String ip,int port){ if(clientHandle!=null) { clientHandle.stop(); } clientHandle = new ClientHandle(ip,port); new Thread(clientHandle,"Server").start(); } //向服务器发送消息 public static boolean sendMsg(String msg) throws Exception{ if(msg.equals("q")) { return false; } clientHandle.sendMsg(msg); return true; } public static void main(String[] args){ start(); } }
package cn.itcast.nio;
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.SocketChannel; import java.util.Iterator; import java.util.Set;
public class ClientHandle implements Runnable{ private String host; private int port; private Selector selector; private SocketChannel socketChannel; private volatile boolean started;
public ClientHandle(String ip,int port) { this.host = ip; this.port = port; try{ //创建选择器 selector = Selector.open(); //打开监听通道 socketChannel = SocketChannel.open(); //如果为 true，则此通道将被置于阻塞模式；如果为 false，则此通道将被置于非阻塞模式 socketChannel.configureBlocking(false);//开启非阻塞模式 started = true; }catch(IOException e){ e.printStackTrace(); System.exit(1); } } public void stop(){ started = false; } @Override public void run() { try{ doConnect(); }catch(IOException e){ e.printStackTrace(); System.exit(1); } //循环遍历selector while(started){ try{ //无论是否有读写事件发生，selector每隔1s被唤醒一次 selector.select(1000); //阻塞,只有当至少一个注册的事件发生的时候才会继续. // selector.select(); Set keys = selector.selectedKeys(); Iterator it = keys.iterator(); SelectionKey key = null; while(it.hasNext()){ key =(SelectionKey) it.next(); it.remove(); try{ handleInput(key); }catch(Exception e){ if(key != null){ key.cancel(); if(key.channel() != null){ key.channel().close(); } } } } }catch(Exception e){ e.printStackTrace(); System.exit(1); } } //selector关闭后会自动释放里面管理的资源 if(selector != null) try{ selector.close(); }catch (Exception e) { e.printStackTrace(); } } private void handleInput(SelectionKey key) throws IOException{ if(key.isValid()){ SocketChannel sc = (SocketChannel) key.channel(); if(key.isConnectable()){ if(sc.finishConnect()); else System.exit(1); } //读消息 if(key.isReadable()){ //创建ByteBuffer，并开辟一个1M的缓冲区 ByteBuffer buffer = ByteBuffer.allocate(1024); //读取请求码流，返回读取到的字节数 int readBytes = sc.read(buffer); //读取到字节，对字节进行编解码 if(readBytes>0){ //将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作 buffer.flip(); //根据缓冲区可读字节数创建字节数组 byte[] bytes = new byte[buffer.remaining()]; //将缓冲区可读字节数组复制到新建的数组中 buffer.get(bytes); String result = new String(bytes,"UTF-8"); System.out.println("客户端收到消息：" + result); } //链路已经关闭，释放资源 else if(readBytes<0){ key.cancel(); sc.close(); } } } } //异步发送消息 private void doWrite(SocketChannel channel,String request) throws IOException{ //将消息编码为字节数组 byte[] bytes = request.getBytes(); //根据数组容量创建ByteBuffer ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length); //将字节数组复制到缓冲区 writeBuffer.put(bytes); //flip操作 writeBuffer.flip(); //发送缓冲区的字节数组 channel.write(writeBuffer); } private void doConnect() throws IOException{ if(socketChannel.connect(new InetSocketAddress(host,port))); else socketChannel.register(selector, SelectionKey.OP_CONNECT); } public void sendMsg(String msg) throws Exception{ socketChannel.register(selector, SelectionKey.OP_READ); doWrite(socketChannel, msg); } }
测试： package cn.itcast.nio;
import java.util.Random; import java.util.Scanner;
public class NIOTest { //测试主方法 public static void main(String[] args) throws Exception{ //运行服务器 Server.start(); //避免客户端先于服务器启动前执行代码 Thread.sleep(100); //运行客户端 //运行客户端 Client.start(); while(Client.sendMsg(new Scanner(System.in).nextLine())){}; } }
5.4. 总结：打开ServerSocketChannel，监听客户端连接 绑定监听端口，设置连接为非阻塞模式 创建Reactor线程，创建多路复用器并启动线程 将ServerSocketChannel注册到Reactor线程中的Selector上，监听ACCEPT事件 Selector轮询准备就绪的key Selector监听到新的客户端接入，处理新的接入请求，完成TCP三次握手，简历物理链路 设置客户端链路为非阻塞模式 将新接入的客户端连接注册到Reactor线程的Selector上，监听读操作，读取客户端发送的网络消息 异步读取客户端消息到缓冲区 对Buffer编解码，处理半包消息，将解码成功的消息封装成Task 将应答消息编码为Buffer，调用SocketChannel的write将消息异步发送给客户端。 因为应答消息的发送，SocketChannel也是异步非阻塞的，所以不能保证一次能吧需要发送的数据发送完，此时就会出现写半包的问题。我们需要注册写操作，不断轮询Selector将没有发送完的消息发送完毕，然后通过Buffer的hasRemain()方法判断消息是否发送完成。
| 
收藏
淘帖
踩
