网络编程之BIO/NIO基础

朴灿烈づ我的快乐病毒、 2022-05-18 22:09 285阅读 0赞

### 什么是网络编程 ###

网络编程是指编写运行在多个设备上（计算机）的程序, 通过网络进行数据交换. 比如现在流行的微服务, 把一个大的系统按照功能拆分多个微服务, 每个微服务都是一个独立的应用, 部署在不同的服务器上, 不同服务器上的微服务如何进行通信就是属于网络编程的范畴.

### TCP、IP、HTTP、Socket的区别 ###

网络模型(OSI)从下往上分为七层, 分别是物理层、数据链路层、网络层、传输层、会话层、表示层和应用层. IP协议是属于网络层的协议, TCP是属于传输层的协议, HTTP是属于应用层的协议, Socket则是对TCP/IP协议的封装和应用.  
![这里写图片描述][70]

### 网络编程三要素 ###

网络编程三要素分别是IP、端口号、TCP/UDP协议. IP是每个设备在网络中的唯一标识, 端口号是每个程序在设备上的唯一标识, TCP/UDP是数据传输的协议.

(1)TCP协议和UDP协议的区别

*  TCP协议是面向连接的(三次握手), 数据安全, 速度慢.
 *  UDP协议是面向无连接的, 数据不安全, 速度快.

(2)TCP协议的三次握手

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70]

*  第一次握手：客户端向服务端发送一个连接报文(标识位SYN=1, 序列号为一个随机值).
 *  第二次握手：服务端收到客户端的连接报文后, 返回一个确认报文(标识位SYN=1和ACK=1, 序列号为一个随机值, 确认号为客户端的序列号 + 1).
 *  第三次握手：客户端收到服务端的确认报文后, 再向服务端发送一个确认报文(标识位ACK=1, 序列号值等于上一次收到的确认号, 确认号为上一次收到的序列号 + 1).

握手过程中传送的包里不包含数据，三次握手完毕后，客户端与服务器才正式开始传送数据。理想状态下，TCP连接一旦建立，在通信双方中的任何一方主动关闭连接之前，TCP 连接都将被一直保持下去。断开连接时服务器和客户端均可以主动发起断开TCP连接的请求，断开过程需要经过“四次挥手”.

(3)TCP协议的四次挥手

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 1]

*  第一次挥手：客户端完成它的数据发送任务后, 向服务端发送一个终止报文, 表示自己不再发送数据. (标识位FIN=1, 序列号值等于上一次收到的确认号)
 *  第二次挥手：服务端收到客户端的终止报文后, 会向客户端返回一个确认报文.(标识位ACK=1, 序列号值等于上一次收到的确认号, 确认号为客户端的序列号 + 1)
 *  第三次挥手：服务端完成它的数据发送任务后, 会向客户端发送一个终止报文, 表示自己不再发送数据.(标识位FIN=1, 序列号值等于上一次收到的确认号, 确认号为第二次挥手的确认号 + 1)
 *  第四次挥手：客户端收到服务端的终止报文后, 会向服务端发送一个确认报文. (标识位ACK=1, 序列号值等于上一次收到的确认号, 确认号为上一次收到的序列号+1)

由于TCP连接是双向的, 因此每个方向都必须单独进行关闭. 当一方完成它的数据发送任务后就会发送一个FIN来终止这个方向的连接, 收到一个FIN只意味着这一方向上没有数据流动, 一个TCP连接在收到一个FIN后仍能发送数据. 首先进行关闭的一方将执行主动关闭, 而另一方执行被动关闭.

(4)为什么建立连接是三次握手, 而关闭连接却是四次握手呢?

这是因为服务端LISTEN状态下的SOCKET当收到SYN报文的连接请求后, 它可以把ACK和SYN放在一个报文里发送. 但关闭连接时, 当收到客户端的FIN报文通知时, 它仅仅表示客户端没有数据发送给服务端, 但未必服务端的数据都发送给客户端了, 所以服务端不会马上关闭SOCKET连接. 当服务端数据发送完毕后, 会发送给客户端一个FIN报文, 表示同意关闭连接, 所以这里的ACK报文和FIN报文是分开发送的.

### BIO(Blocking IO) ###

BIO也叫同步阻塞IO, 对于每一个客户端的连接请求都会创建一个新线程来进行处理, 处理完成后线程销毁. 当一个线程调用IO流读写数据时，该线程被阻塞，直到读到数据，或数据完全写入, 该线程在此期间不能再干任何事情。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 2]

public class Server {
    	public static void main(String[] args) {
    		ServerSocket server = null;
    		try {
    			server = new ServerSocket(12000);
    			System.out.println("server start...");
    			while(true){
    				//进行阻塞,监听端口
    				Socket socket = server.accept();
    				//新建一个线程执行客户端的任务
    				new Thread(new ServerHandler(socket)).start();
    			}
    			
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			if(server != null){
    				try {
    					server.close();
    				} catch (IOException e) {
    					e.printStackTrace();
    				}
    			}
    			server = null;
    		}
    	}
    }
    
    public class ServerHandler implements Runnable{
    	private Socket socket ;
    	
    	public ServerHandler(Socket socket){
    		this.socket = socket;
    	}
    	
    	@Override
    	public void run() {
    		BufferedReader in = null;
    		PrintWriter out = null;
    		try {
    			in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
    			out = new PrintWriter(this.socket.getOutputStream(), true);
    			String content = null;
    			while(true){
    				content = in.readLine();
    				if(content == null) break;
    				System.out.println("Server :" + content);
    				out.println("Server response");
    			}
    			
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			if(in != null){
    				try {
    					in.close();
    				} catch (IOException e) {
    					e.printStackTrace();
    				}
    			}
    			if(out != null){
    				try {
    					out.close();
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    			if(socket != null){
    				try {
    					socket.close();
    				} catch (IOException e) {
    					e.printStackTrace();
    				}
    			}
    			socket = null;
    		}
    	}
    }
    
    public class Client {
    	public static void main(String[] args) {
    		Socket socket = null;
    		BufferedReader in = null;
    		PrintWriter out = null;
    		
    		try {
    			socket = new Socket("127.0.0.1", 12000);
    			in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
    			out = new PrintWriter(socket.getOutputStream(), true);
    			
    			//向服务器端发送数据
    			out.println("Client request");
    			String response = in.readLine();
    			System.out.println("Client: " + response);
    			
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			if(in != null){
    				try {
    					in.close();
    				} catch (IOException e) {
    					e.printStackTrace();
    				}
    			}
    			if(out != null){
    				try {
    					out.close();
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    			if(socket != null){
    				try {
    					socket.close();
    				} catch (IOException e) {
    					e.printStackTrace();
    				}
    			}
    			socket = null;
    		}
    	}
    }

**伪异步IO**

使用线程池来管理线程, 实现1个或多个线程处理N个客户端.

public class Server {
    	public static void main(String[] args) {
    		ServerSocket server = null;
    		try {
    			server = new ServerSocket(12000);
    			System.out.println("server start...");
    			while(true){
    				//进行阻塞,监听端口
    				Socket socket = server.accept();
    				
    				//用线程池来管理线程
    				ThreadPoolExecutor executor = new ThreadPoolExecutor(
    						Runtime.getRuntime().availableProcessors(), 
    						10, 
    						120L, 
    						TimeUnit.SECONDS, 
    						new LinkedBlockingQueue<Runnable>(20)
    						);
    				
    				//新建一个线程执行客户端的任务
    				executor.execute(new Thread(new ServerHandler(socket)));
    			}
    			
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			if(server != null){
    				try {
    					server.close();
    				} catch (IOException e) {
    					e.printStackTrace();
    				}
    			}
    			server = null;
    		}
    	}
    }

### NIO(Non-Blocking IO) ###

NIO也叫同步非阻塞IO, NIO的服务端可以只启动一个专门的线程来处理所有的 IO 事件, 且不会被任何IO事件阻塞住.

服务端和客户端各自维护一个Selector选择器, Selector会不断地轮询注册在其上的通道(Channel)是否发生事件, 只有事件发生时才会去执行相应的操作. 当客户端连接服务器时发生OP\_CONNECT事件, 当服务端接收到客户端连接时发生OP\_ACCEPT事件, 当有数据发送过来时发生OP\_READ事件, 当要发送数据给对方时发生OP\_WRITE事件.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 3]

**Buffer**

在BIO中, 数据直接读写到Stream对象中. 而在NIO中, 所有数据都是读写到Buffer中, 然后通过Channel传输. Buffer实质上是一个数组, 通常它是一个字节数组(ByteBuffer) ,也可以是其他类型的数组.

(1)成员变量

*  mark : s初始值为-1，用于备份当前的position;
 *  position : 初始值为0, position表示当前可以写入或读取数据的位置，当写入或读取一个数据后，position移动到下一个位置
 *  limit : 写模式下，limit表示最多能往Buffer里写多少数据，等于capacity值；读模式下，limit表示最多可以读取多少数据
 *  capacity : 缓存数组大小

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 4]

(2)成员方法

*  allocate(int capacity) : 创建指定长度的缓冲区
 *  put(E e) : 添加一个元素
 *  get() : 获取第一个元素
 *  wrap(E\[\] array) : 将数组元素添加到缓冲中
 *  clear() : 清除数据, 实际上数据没有被清除.
 *

public final Buffer clear() {
            position = 0;
            limit = capacity;
            mark = -1;
            return this;
        }

*  flip() : Buffer有两种模式, 写模式和读模式. flip后Buffer从写模式变成读模式.
 *

public final Buffer flip() {
            limit = position;
            position = 0;
            mark = -1;
            return this;
        }

(3)Buffer的实现类

*  ByteBuffer
 *  CharBuffer
 *  ShortBuffer
 *  IntBuffer
 *  LongBuffer
 *  FloatBuffer
 *  DoubleBuffer

(4)实例

public static void main(String[] args) {
    	//创建指定长度的缓冲区
    	IntBuffer buf = IntBuffer.allocate(10);
    	buf.put(13);
    	buf.put(21);
    	buf.put(35);
    	System.out.println(buf);
    	
    	//从写模式变成读模式
    	buf.flip();
    	System.out.println(buf);
    	
    	//调用get方法会使position位置向后递增一位
    	for (int i = 0; i < buf.limit(); i++) {
    		System.out.print(buf.get() + "\t");
    	}
    	System.out.println("\n" + buf);
    	
    	//清除数据
    	buf.clear();	
    	System.out.println(buf);
    }
    
    java.nio.HeapIntBuffer[pos=3 lim=10 cap=10]
    java.nio.HeapIntBuffer[pos=0 lim=3 cap=10]
    13	21	35	
    java.nio.HeapIntBuffer[pos=3 lim=3 cap=10]
    java.nio.HeapIntBuffer[pos=0 lim=10 cap=10]

**Channel**

NIO把它支持的I/O对象抽象为Channel, Channel又称为"通道", 类似于BIO中的流(Stream), 但有锁区别:

*  流是单向的，通道是双向的，可读可写
 *  流读写是阻塞的，通道可以阻塞也可以非阻塞
 *  流中的数据可以选择性的先读到缓存中，通道的数据总是要先读写到缓存中

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 5]

(1)Channel的实现类

*  FileChannel
 *  DatagramChannel
 *  SocketChannel
 *  ServerSocketChannel

**Selector**

Selector选择器充当一个监听者, 会不断地轮询注册在其上的通道(Channel)是否发生事件, 如果某个通道发生了读写操作, 这个通道就处于就绪状态, 会被Selector轮询出来, 进行后续的IO操作.

(1)Selector监听的事件(SelectionKey)

*  OP\_CONNECT : 客户端连接服务端事件
 *  OP\_ACCEPT : 服务端接收客户端连接事件
 *  OP\_READ : 读事件
 *  OP\_WRITE : 写事件

**简单的NIO实例**

public class Server{
    	//选择器(多路复用器)
    	private Selector selector;
    	//读取的缓冲区
    	private ByteBuffer readBuf = ByteBuffer.allocate(1024);
    	//写入的缓冲区
    	private ByteBuffer writeBuf = ByteBuffer.allocate(1024);
    	
    	/**
    	 * 打开选择器,注册服务器通道
    	 */
    	public Server(int port){
    		try {
    			//1 打开选择器
    			this.selector = Selector.open();
    			//2 打开服务器通道
    			ServerSocketChannel ssc = ServerSocketChannel.open();
    			//3 设置服务器通道为非阻塞模式
    			ssc.configureBlocking(false);
    			//4 绑定地址
    			ssc.bind(new InetSocketAddress(port));
    			// 5 把服务器通道注册到选择器上,并为该通道注册OP_ACCEPT事件. 
    			// 当该事件到达时,selector.select()会返回,否则selector.select()会一直阻塞	
    			ssc.register(this.selector, SelectionKey.OP_ACCEPT);
    			
    			System.out.println("Server start, port :" + port);
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    	}
    	
    	/**
    	 * 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理
    	 */
    	public void listen() {
    		while(true){
    			try {
    				//1 当注册的事件发生时,方法返回;否则,该方法会一直阻塞
    				this.selector.select();
    				
    				//2 获得selector选中项的迭代器，选中的项为注册的事件
    				Iterator<SelectionKey> keys = this.selector.selectedKeys().iterator();
    				while(keys.hasNext()){
    					SelectionKey key = keys.next();
    					//删除已选的key,以防重复处理 
    					keys.remove();
    					
    					//OP_ACCEPT事件发生(接收到客户端的连接)
    					if(key.isAcceptable()){
    						ServerSocketChannel ssc =  (ServerSocketChannel) key.channel();
    						//获得和客户端连接的通道  
    						SocketChannel sc = ssc.accept();
    						sc.configureBlocking(false);
    						//将SocketChannel注册到selector上,并设置读事件
    						sc.register(this.selector, SelectionKey.OP_READ);
    					}else if(key.isReadable()){ //OP_READ事件发生
    						read(key);
    					}else if(key.isWritable()){ //OP_WRITE事件发生
    						write(key);
    					}
    				}
    			} catch (IOException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    	/**
    	 * 读操作
    	 */
    	private void read(SelectionKey key) {
    		try {
    			//清空缓冲区旧的数据
    			this.readBuf.clear();
    			//获取socket通道对象
    			SocketChannel sc = (SocketChannel) key.channel();
    			int count = sc.read(this.readBuf);
    				
    			if(count > 0){
    				String msg = new String(readBuf.array());  
    				System.out.println("服务端收到信息："+msg);  
    				
    				//将SocketChannel注册到selector上,并设置写事件
    				sc.register(this.selector, SelectionKey.OP_WRITE);
    			}
    			
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    	}
    	
    	/**
    	 * 写操作
    	 */
    	private void write(SelectionKey key){
    		try {
    			//清空缓冲区旧的数据
    			this.writeBuf.clear();
    			//获取socket通道对象
    			SocketChannel sc = (SocketChannel) key.channel();
    			sc.write(ByteBuffer.wrap(new String("我是服务端,我已收到你的信息!").getBytes()));
    			
    			//将SocketChannel注册到selector上,并设置读事件
    			sc.register(this.selector, SelectionKey.OP_READ);
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    	}
    	
    	public static void main(String[] args) {
    		new Server(12000).listen();
    	}
    }
    
    public class Client {
    	//选择器(多路复用器)
    	private Selector selector;
    	//读取的缓冲区
    	private ByteBuffer readBuf = ByteBuffer.allocate(1024);
    	//写入的缓冲区
    	private ByteBuffer writeBuf = ByteBuffer.allocate(1024);
    	
    	/**
    	 * 打开选择器,注册客户端通道
    	 */
    	public Client(String ip,int port){
    		try {
    			//1 打开选择器
    			this.selector = Selector.open();  
    	        //2 获得一个Socket通道  
    	        SocketChannel channel = SocketChannel.open();  
    	        //3 设置通道为非阻塞  
    	        channel.configureBlocking(false);  
    	        //4 绑定服务器ip和port
    	        channel.connect(new InetSocketAddress(ip,port));  
    	        //5 将客户端通道注册到选择器上，并为该通道注册OP_CONNECT事件	
    	        channel.register(selector, SelectionKey.OP_CONNECT);  
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    	}
    	
    	/**
    	 * 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理
    	 */
    	public void listen() {
    		while(true){
    			try {
    				//1 当注册的事件发生时,方法返回;否则,该方法会一直阻塞
    				this.selector.select();
    				
    				//2 获得selector选中项的迭代器，选中的项为注册的事件
    				Iterator<SelectionKey> keys = this.selector.selectedKeys().iterator();
    				while(keys.hasNext()){
    					SelectionKey key = keys.next();
    					//删除已选的key,以防重复处理 
    					keys.remove();
    					
    					//OP_CONNECT事件发生(连接上服务器)
    					if(key.isConnectable()){
    						SocketChannel sc =  (SocketChannel) key.channel();
    						// 如果正在连接，则完成连接  
    	                    if(sc.isConnectionPending()){  
    	                    	sc.finishConnect();  
    	                    }  
    						sc.configureBlocking(false);
    						
    						//将SocketChannel注册到selector上,并设置写事件
    						sc.register(this.selector, SelectionKey.OP_WRITE);
    					}else if(key.isReadable()){
    						read(key);
    					}else if(key.isWritable()){
    						write(key);
    					}
    				}
    			} catch (IOException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    	/**
    	 * 读操作
    	 */
    	private void read(SelectionKey key) {
    		try {
    			//清空缓冲区旧的数据
    			this.readBuf.clear();
    			//获取socket通道对象
    			SocketChannel sc = (SocketChannel) key.channel();
    			int count = sc.read(this.readBuf);
    				
    			if(count > 0){
    				String msg = new String(readBuf.array());  
    				System.out.println("客户端收到信息："+msg);  
    				
    				//将SocketChannel注册到selector上,并设置写事件
    				sc.register(this.selector, SelectionKey.OP_WRITE);
    			}
    			
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    	}
    	
    	
    	/**
    	 * 写操作
    	 */
    	private void write(SelectionKey key){
    		try {
    			//清空缓冲区旧的数据
    			this.writeBuf.clear();
    			//获取socket通道对象
    			SocketChannel sc = (SocketChannel) key.channel();
    			sc.write(ByteBuffer.wrap(new String("我是Client,我先发条信息!").getBytes()));
    			
    			//将SocketChannel注册到selector上,并设置读事件
    			sc.register(this.selector, SelectionKey.OP_READ);
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    	}
    	
    	public static void main(String[] args) {
    		new Client("127.0.0.1",12000).listen();
    	}
    }
    
    结果:
    
    客户端先发送一条消息
    服务端收到客户端的消息,然后返回一条消息
    客户端收到服务端的消息,再发送一条消息
    ...

### BIO和NIO的区别 ###

(1)BIO是面向流的, 而NIO是面向缓冲的.

(2)BIO是阻塞的,而NIO是非阻塞的.

*  传统IO方式(BIO)在调用InputStream.read()/BufferedReader.readLine()方法时是阻塞的，它会一直等到数据到来或缓冲区已满或超时才会返回.
 *  NIO通过向Selector注册读写事件, Selector不断轮询读写事件是否发生, 当读写事件发生后再去进行相应的处理.

(3)NIO的选择器允许一个单独的线程来监视多个输入通道.

[70]: /images/20220519/714c8df7badc40fe92bb80699e10a2ed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70]: /images/20220519/8be385d3967445c3ad1cbd96446b44e9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 1]: /images/20220519/0a5409704e584b47ad474402be61942d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 2]: /images/20220519/68082b9a08604a6f864a250801a0c3f3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 3]: /images/20220519/b71bf8845f0241d49da93c9fd1d7905d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 4]: /images/20220519/cecab832222544448b41ccbe69dfbc49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70 5]: /images/20220519/d9c8480fa4fd4a379f7ada22987cce1c.png