分布式之Netty介绍与实战(一)--Java IO与NIO

缺乏、安全感 2022-05-14 16:25 216阅读 0赞

[分布式之Netty介绍与实战(一)--Java IO与NIO][Netty_--Java IO_NIO]

[分布式之Netty介绍与实战(二)--Netty编程实践][Netty_--Netty]

[分布式之Netty介绍与实战(三)--Netty线程模型解析][Netty_--Netty 1]

[分布式之Netty介绍与实战(四)--Netty编解码编程实战][Netty_--Netty 2]

--------------------

# Linux IO流程 #

# ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70][] #

## lLinux IO流程： ##

等待数据准备好（Waiting for the data to be ready）

从内核向进程复制数据（Copying the data from the kernel to the process）

# Linux IO模型 #

## l《Unix网络编程》 ##

UNIX网络编程的经典之作

## l将I/O模型划分为以下五种类型： ##

#  阻塞式I/O模型 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 1][]

# 非阻塞式I/O模型 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 2][]

# I/O复用模型 #

一次观察多个I/O进程，等待哪个进程好了就调用哪个进程

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 3][]

# 信号驱动式I/O模型 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 4][]

# 异步I/O模型 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 5][]

# 各种I/O模型的比较 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 6][]

# 同步 vs 异步 #

*  阻塞式I/O模型
 *  非阻塞式I/O模型
 *  I/O复用
 *  信号驱动式I/O
 *  异步I/O

## lPOSIX标准将同步I/O和异步I/O定义为： ##

同步I/O操作：导致请求进程阻塞，直到I/O操作完成。

# 异步I/O操作：不导致请求进程阻塞。   Blocking I/O #

#  Single Thread  Server （阻塞模式）   #

public class SingleThreadEchoServer {
    
    	private int port;
    
    	public SingleThreadEchoServer(int port) {
    		this.port = port;
    	}
    
    	public void startServer() {
    		ServerSocket echoServer = null;
    		int i = 0;
    		System.out.println("服务器在端口[" + this.port + "]等待客户请求......");
    		try {
    			echoServer = new ServerSocket(this.port);
    			while (true) {
    				Socket clientRequest = echoServer.accept();
    				handleRequest(clientRequest, i++);
    			}
    		} catch (IOException e) {
    			System.out.println(e);
    		}
    	}
    
    	private void handleRequest(Socket clientSocket, int clientNo) {
    		PrintStream os = null;
    		BufferedReader in = null;
    		try {
    			in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
    			os = new PrintStream(clientSocket.getOutputStream());
    			String inputLine;
    			while ((inputLine = in.readLine()) != null) {
    
    				// 输入'Quit'退出
    				if (inputLine.equals("Quit")) {
    					System.out.println("关闭与客户端[" + clientNo + "]......" + clientNo);
    					os.close();
    					in.close();
    					clientSocket.close();
    					break;
    				} else {
    					System.out.println("来自客户端[" + clientNo + "]的输入： [" + inputLine + "]！");
    					os.println("来自服务器端的响应：" + inputLine);
    				}
    			}
    		} catch (IOException e) {
    			System.out.println("Stream closed");
    		}
    	}
    
    	public static void main(String[] args) throws IOException {
    		new SingleThreadEchoServer(8030).startServer();
    	}
    }

# MultiThread Server （非阻塞模式） #

### 新来一个就新建一个线程 ###

## ##

public class MultiThreadedEchoServerV1 {
    	private int port;
    	
    	public MultiThreadedEchoServerV1(int port) {
    		this.port = port;
    	}
    	
    	public void startServer() {
     		ServerSocket echoServer = null;
    		int i = 0;
    		System.out.println("服务器在端口[" + this.port + "]等待客户请求......");
    		try {
    			echoServer = new ServerSocket(port);
    			while (true) {
    				Socket clientRequest = echoServer.accept();
    				new Thread(new ThreadedServerHandler(clientRequest, i++)).start();
    			}
    		} catch (IOException e) {
    			System.out.println(e);
    		}
    
    	}
     
        public static void main(String[] args) throws IOException {
        	new MultiThreadedEchoServerV1(8030).startServer();
         }
    }

public class ThreadedServerHandler implements Runnable {
    	Socket clientSocket = null;
    	int clientNo = 0;
    
    	ThreadedServerHandler(Socket socket, int i) {
    		if (socket != null) {
    			clientSocket = socket;
    			clientNo = i;
    			System.out.println("创建线程为[" + clientNo + "]号客户服务...");
    		}
    	}
    
    	@Override
    	public void run() {
    		PrintStream os = null;
    		BufferedReader in = null;
    		try {
    			in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
    			os = new PrintStream(clientSocket.getOutputStream());
    			String inputLine;
    			while ((inputLine = in.readLine()) != null) {
    
    				// 输入'Quit'退出
    				if (inputLine.equals("Quit")) {
    					System.out.println("关闭与客户端[" + clientNo + "]......" + clientNo);
    					os.close();
    					in.close();
    					clientSocket.close();
    					break;
    				} else {
    					System.out.println("来自客户端[" + clientNo + "]的输入： [" + inputLine + "]！");
    					os.println("来自服务器端的响应：" + inputLine);
    				}
    			}
    		} catch (IOException e) {
    			System.out.println("Stream closed");
    		}
    	}
    }

### 非阻塞模式升级，增加线程池处理，线程交付线程池管理 ###

### ![20181028203953259.png][] ###

public class MultiThreadedEchoServerV2 {
    	private int port;
    
    	public MultiThreadedEchoServerV2(int port) {
    		this.port = port;
    	}
    
    	public void startServer() {
    		ServerSocket echoServer = null;
    		Executor executor = Executors.newFixedThreadPool(5);
    		int i = 0;
    		System.out.println("服务器在端口[" + this.port + "]等待客户请求......");
    		try {
    			echoServer = new ServerSocket(8080);
    			while (true) {
    				Socket clientRequest = echoServer.accept();
    				executor.execute(new ThreadedServerHandler(clientRequest, i++));
    			}
    		} catch (IOException e) {
    			System.out.println(e);
    		}
    	}
    
    	public static void main(String[] args) throws IOException {
    		new MultiThreadedEchoServerV2(8080).startServer();
    
    	}
    }

*  ServerSocket
 *  Socket  
    ![20181028203342817.png][]

# Java NIO概览——变迁 #

*  NIO = New I/O
 *  NIO 1: JSR 51
    
     *  JDK 1.4引入
     *  http://jcp.org/en/jsr/detail?id=051

NIO 2: JSR203

*  JDK 7
 *  http://jcp.org/en/jsr/detail?id=203

# Java IO vs NIO #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 7][]

# Java NIO #

*  NIO
    
     *  Buffers
     *  Channels
     *  Selectors
 *  NIO 2.0
    
     *  Update
     *  New File System API
     *  Asynchronous I/O

Java NIO 主要由下面3部分核心组件组成：

*  Buffer
 *  Channel
 *  Selector

# Java NIO Buffer #

Java NIO Buffer

*  一个 Buffer 本质上是内存中的一块， 可以将数据写入这块内存， 从这块内存获取数据 
 *  java.nio 定义了以下几个 Buffer 的实现

![2018102820495037.png][]

Java NIO Buffer三大核心概念：position、limit、capacity

*  最好理解的当然是 capacity，它代表这个缓冲区的容量，一旦设定就不可以更改。比如 capacity 为 1024 的 IntBuffer，代表其一次可以存放 1024 个 int 类型的值。
 *  一旦 Buffer 的容量达到 capacity，需要清空 Buffer，才能重新写入值。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 8][]

从写操作模式到读操作模式切换的时候（flip），position 都会归零，这样就可以从头开始读写了。

写操作模式下，limit 代表的是最大能写入的数据，这个时候 limit 等于 capacity。

写结束后，切换到读模式，此时的 limit 等于 Buffer 中实际的数据大小，因为 Buffer 不一定被写满了。

![20181028205144568.png][]

java.nio.buffer,缓冲区抽象

## ByteBuffer ##

*  理解capacity、limit、position、mark
 *  0 – mark – position – limit – capacity

Direct ByteBuffer VS. non-direct ByteBuffer

### Non-direct ByteBuffer ###

*  HeapByteBuffer,标准的java类
 *  维护一份byte\[\]在JVM堆上
 *  创建开销小

### Direct ByteBuffer ###

* 底层存储在非JVM堆上,通过native代码操作
 * \-XX:MaxDirectMemorySize=<size>
 * 创建开销大

### 两种类型的ByteBuffer ###

<table> 
 <tbody> 
 <tr> 
 <td style="background-color:#4f81bd;border-color:#ffffff;vertical-align:top;width:169pt;"> &nbsp; </td> 
 <td style="background-color:#4f81bd;border-color:#ffffff;text-align:center;vertical-align:middle;width:309pt;"> DirectByteBuffer </td> 
 <td style="background-color:#4f81bd;border-color:#ffffff;text-align:center;vertical-align:middle;width:337pt;"> HeapByteBuffer </td> 
 </tr> 
 <tr> 
 <td style="background-color:#d0d8e8;border-color:#ffffff;vertical-align:middle;width:169pt;"> 创建开销&nbsp;&nbsp;&nbsp; </td> 
 <td style="background-color:#d0d8e8;border-color:#ffffff;vertical-align:middle;width:309pt;"> 大 </td> 
 <td style="background-color:#d0d8e8;border-color:#ffffff;vertical-align:middle;width:337pt;"> 小 &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="background-color:#e9edf4;border-color:#ffffff;vertical-align:middle;width:169pt;"> 存储位置 </td> 
 <td style="background-color:#e9edf4;border-color:#ffffff;vertical-align:middle;width:309pt;"> Native heap </td> 
 <td style="background-color:#e9edf4;border-color:#ffffff;vertical-align:middle;width:337pt;"> &nbsp;Java heap &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="background-color:#d0d8e8;border-color:#ffffff;vertical-align:middle;width:169pt;"> 数据拷贝 </td> 
 <td style="background-color:#d0d8e8;border-color:#ffffff;vertical-align:middle;width:309pt;"> 无需临时缓冲区做拷贝 </td> 
 <td style="background-color:#d0d8e8;border-color:#ffffff;vertical-align:middle;width:337pt;"> 拷贝到临时DirectByteBuffer,但临时缓冲区使用缓存。 聚集写/发散读时没有缓存临时缓冲区。 </td> 
 </tr> 
 <tr> 
 <td style="background-color:#e9edf4;border-color:#ffffff;vertical-align:middle;width:169pt;"> GC影响 </td> 
 <td style="background-color:#e9edf4;border-color:#ffffff;vertical-align:middle;width:309pt;"> 每次创建或者释放的时候 都调用一次System.gc() </td> 
 <td style="background-color:#e9edf4;border-color:#ffffff;vertical-align:middle;width:337pt;"> &nbsp; </td> 
 </tr> 
 </tbody> 
</table>

# Buffer Basic Operations #

*  Creating
    
     *  allocate/allocateDirect方法
     *  warp
 *  Filling and Draining
 *  Flipping and Rewind
 *  Marking
 *  Comparing
 *  Duplication

# Java NIO Channel #

所有的 NIO 操作始于通道，通道是数据来源或数据写入的目的地，主要地， java.nio 包中主要实现的以下几个 Channel：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 9][]

*  FileChannel：文件通道，用于文件的读和写
 *  DatagramChannel：用于 UDP 连接的接收和发送
 *  SocketChannel：把它理解为 TCP 连接通道，简单理解就是 TCP 客户端
 *  ServerSocketChannel：TCP 对应的服务端，用于监听某个端口进来的请求

![20181028210955705.png][]

# Java NIO Selector #

Selector是Java NIO中的一个组件，用于检查一个或多个NIO Channel的状态是否处于可读、可写

如此可以实现单线程管理多个channels,也就是可以管理多个网络链接

![2018102821105491.png][]

*  java.nio.channels.Selector
 *  支持IO多路复用的抽象实体
 *  注册Seletable Channel
 *  SelectionKey —— 表示Selector和被注册的channel之间关系,一份凭证
    
     *  SelectionKey 保存channel感兴趣的事件
 *  Selector.select 更新所有就绪的SelectionKey的状态,并返回就绪的channel个数
    
     *  迭代Selected Key集合并处理就绪channel

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 10][]

创建Selector(Creating a Selector)

Selector selector = Selector.open();

注册Channel到Selector上(Registering Channels with the Selector)

channel.configureBlocking(false);
    SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

register的第二个参数，这个参数是一个“关注集合”，代表关注的channel状态，

有四种基础类型可供监听, 用SelectionKey中的常量表示如下：

SelectionKey.OP_CONNECT
    SelectionKey.OP_ACCEPT
    SelectionKey.OP_READ
    SelectionKey.OP_WRITE

从Selector中选择channel(Selecting Channels via a Selector)

* 一旦向Selector注册了一个或多个channel后，就可以调用select来获取channel
 * select方法会返回所有处于就绪状态的channel
 * select方法具体如下：
 
 * int select()
 * int select(long timeout)
 * int selectNow()
 * select()方法的返回值是一个int，代表有多少channel处于就绪了。也就是自上一次select后有多少channel进入就绪。
 * selectedKeys()
 
 * 在调用select并返回了有channel就绪之后，可以通过选中的key集合来获取channel，
 * 这个操作通过调用selectedKeys()方法：
 
 Set<SelectionKey> selectedKeys = selector.selectedKeys();
 
 Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
 
 while(keyIterator.hasNext()) {
 
 SelectionKey key = keyIterator.next();
 
 if(key.isAcceptable()) {
 // a connection was accepted by a ServerSocketChannel.
 
 } else if (key.isConnectable()) {
 // a connection was established with a remote server.
 
 } else if (key.isReadable()) {
 // a channel is ready for reading
 
 } else if (key.isWritable()) {
 // a channel is ready for writing
 }
 
 keyIterator.remove();
 }

# NIO带来了什么 #

事件驱动模型

*  避免多线程
 *  单线程处理多任务

非阻塞IO,IO读写不再阻塞,而是返回0

基于block的传输,通常比基于流的传输更高效

更高级的IO函数,zero-copy

IO多路复用大大提高了java网络应用的可伸缩性和实用性

# NIO Tips #

* •使用NIO = 高性能
 
 * NIO不一定更快的场景
 
 * 客户端应用
 * 连接数<1000
 * 并发程度不高
 * 局域网环境下
 * •NIO完全屏蔽了平台差异
 
 * NIO仍然是基于各个OS平台的IO系统实现的,差异仍然存在
 * •使用NIO做网络编程很容易
 
 * 离散的事件驱动模型，编程困难
 * **陷阱重重**
 * **Netty****简化****NIO****编程**

[Netty_--Java IO_NIO]: https://blog.csdn.net/z785495295/article/details/83477789
[Netty_--Netty]: https://blog.csdn.net/z785495295/article/details/83650052
[Netty_--Netty 1]: https://blog.csdn.net/z785495295/article/details/83660993
[Netty_--Netty 2]: https://blog.csdn.net/z785495295/article/details/83721048
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70]: /images/20220503/23c20718362342da9a757ee3b94f3a4e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 1]: /images/20220503/923a5f1f265c4c0f8eea65cb3955213d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 2]: /images/20220503/6da692ccfb4a4427b175441ed7d3b71e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 3]: /images/20220503/a7bbc6b6934e4765afc00540abc2ab94.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 4]: /images/20220503/cf44ebd23a8041088b1f323536326d58.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 5]: /images/20220503/8cef6eea8c56406c9dba585e65ad1c26.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 6]: /images/20220503/e9c2d977ea38480b94b4f518fb5cad8a.png
[20181028203953259.png]: /images/20220503/ad4e48052426484aba3819a0fc44e45d.png
[20181028203342817.png]: /images/20220503/c916ce6b844c4ad7aa907776414223ba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 7]: /images/20220503/fbc15be1f87847ce827850a9ced60f4f.png
[2018102820495037.png]: /images/20220503/6832e7add55044a2822e524d75f773f1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 8]: /images/20220503/9616c5b92f8c4ac6a556f41d1f5ab871.png
[20181028205144568.png]: /images/20220503/91214c4ac98344a6a3fe9cc62f35f87a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 9]: /images/20220503/e895eb96fd6c45e0b31c6dd14023fe2c.png
[20181028210955705.png]: /images/20220503/5ffd1b4332b34726900d2c4282ae563a.png
[2018102821105491.png]: /images/20220503/cd9afeef298d4824a77ae7513c2de726.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_27_color_FFFFFF_t_70 10]: /images/20220503/cb861a7bc1b14b169cbb4fffd0442347.png