Java网络编程与NIO

雨点打透心脏的1/2处 2022-04-13 07:53 310阅读 0赞

# 1. 计算机网络编程基础 #

## 1.七层模型 ##

七层模型（OSI，Open System Interconnection参考模型），是参考是国际标准化组织制定的一个用于计算机或通信系统间互联的标准体系。它是一个七层抽象的模型，不仅包括一系列抽象的术语和概念，也包括具体的协议。 经典的描述如下：

简述每一层的含义：

> 1.  物理层（Physical Layer）：建立、维护、断开物理连接。
> 2.  数据链路层 (Link)：逻辑连接、进行硬件地址寻址、差错校验等。
> 3.  网络层 (Network)：进行逻辑寻址，实现不同网络之间的路径选择。
> 4.  传输层 (Transport)：定义传输数据的协议端口号，及流控和差错校验。
> 5.  会话层（Session Layer）：建立、管理、终止会话。
> 6.  表示层（Presentation Layer）：数据的表示、安全、压缩。
> 7.  应用层 (Application)：网络服务与最终用户的一个接口

每一层利用下一层提供的服务与对等层通信，每一层使用自己的协议。了解了这些，然并卵。但是，这一模型确实是绝大多数网络编程的基础，作为抽象类存在的，而TCP／IP协议栈只是这一模型的一个具体实现。

## 2.TCP/IP协议模型 ##

IP数据包结构：  
![IP数据包结构][IP]  
\---  
TCP数据包结构：  
![TCP数据包结构][TCP]

### 一个模型例子： ###

寻址过程：每台机子都有个物理地址MAC地址和逻辑地址IP地址，物理地址用于底层的硬件的通信，逻辑地址用于上层的协议间的通信。寻址过程会先使用ip地址进行路由寻址，在不同网络中进行路由转发，到了同一个局域网时，再根据物理地址进行广播寻址，数据在以太网的局域网中都是以广播方式传输的，整个局域网中的所有节点都会收到该帧，只有目标MAC地址与自己的MAC地址相同的帧才会被接收。

建立可靠的连接：A向B传输一个文件时，如果文件中有部分数据丢失，就可能会造成在B上无法正常阅读或使用。 TCP协议就是建立了可靠的连接：  
**TCP三次握手确定了双方数据包的序号、最大接受数据的大小(window)以及MSS(Maximum Segment Size)**

会话层用来建立、维护、管理应用程序之间的会话，主要功能是对话控制和同步，编程中所涉及的session是会话层的具体体现。表示层完成数据的解编码，加解密，压缩解压缩等。

# 2.Socket编程 #

在Linux世界，“一切皆文件”，操作系统把网络读写作为IO操作，就像读写文件那样，对外提供出来的编程接口就是Socket。所以，socket（套接字）是通信的基石，是支持TCP/IP协议网络通信的基本操作单元。socket实质上提供了进程通信的端点。进程通信之前，双方首先必须各自创建一个端点，否则是没有办法建立联系并相互通信的。一个完整的socket有一个本地唯一的socket号，这是由操作系统分配的。

在许多操作系统中，Socket描述符和其他IO描述符是集成在一起的，操作系统把socket描述符实现为一个指针数组，这些指针指向内部数据结构。进程进行Socket操作时，也有着多种处理方式，如阻塞式IO，非阻塞式IO，多路复用(select/poll/epoll)，AIO等等。  
**多路复用**往往在提升性能方面有着重要的作用。  
当前主流的Server侧Socket实现大都采用了epoll的方式，例如Nginx， 在配置文件可以显式地看到 use epoll。

举个栗子  
Java中Socket服务端的简单实现：基本思路就是一个大循环不断监听客户端请求，为了提高处理效率可以使用线程池多个线程进行每个连接的数据读取

public class BIOServer {
        private ServerSocket serverSocket;
        private ExecutorService executorService = Executors.newCachedThreadPool();
    
        class Handler implements Runnable {
    
            Socket socket;
            
            public Handler(Socket socket) {
               this.socket = socket;
            }
    
            @Override
            public void run() {
                try {
                    BufferedReader buf = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    String readData = buf.readLine();
                    while (readData != null) {
                        readData = buf.readLine();
                        System.out.println(readData);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    
        public BIOServer(int port) {
            try {
                serverSocket = new ServerSocket(port);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        public void run() {
            try {
                Socket socket = serverSocket.accept();
                executorService.submit(new Handler(socket));
            } catch (Exception e) {
    
            }
        }
    }

客户端：建立socket连接、发起请求、读取响应

public class IOClient {
    
        public void start(String host, int port) {
            try {
                Socket s = new Socket("127.0.0.1",8888);
                InputStream is = s.getInputStream();
                OutputStream os = s.getOutputStream();
    
                BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(os));
                bw.write("测试客户端和服务器通信，服务器接收到消息返回到客户端\n");
                bw.flush();
    
                BufferedReader br = new BufferedReader(new InputStreamReader(is));
                String mess = br.readLine();
                System.out.println("服务器："+mess);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

# 3.IO模型 #

对于一次IO访问（以read举例），数据会先被拷贝到操作系统内核的缓冲区page cache中，然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。所以说，当一个read操作发生时，它会经历两个阶段：

1.  等待数据准备
2.  将数据从内核拷贝到进程中

IO模型的分类有下：

*  阻塞 I/O（blocking IO）
 *  非阻塞 I/O（nonblocking IO）
 *  I/O 多路复用（ IO multiplexing）
 *  异步 I/O（asynchronous IO）

## BIO 阻塞 I/O ##

缺点：一个请求一个线程，浪费线程，且上下文切换开销大；

上面写的socket列子就是典型的BIO  
![bio][]

当用户进程调用了recvfrom这个系统调用，kernel就开始了IO的第一个阶段：准备数据（对于网络IO来说，很多时候数据在一开始还没有到达。比如，还没有收到一个完整的UDP包。这个时候kernel就要等待足够的数据到来）。这个过程需要等待，也就是说数据被拷贝到操作系统内核的缓冲区中是需要一个过程的。而在用户进程这边，整个进程会被阻塞（当然，是进程自己选择的阻塞）。当kernel一直等到数据准备好了，它就会将数据从kernel中拷贝到用户内存，然后kernel返回结果，用户进程才解除block的状态，重新运行起来。

## NIO 非阻塞 I/O ##

![bio1][]  
当用户进程发出read操作时，**如果kernel中的数据还没有准备好，那么它并不会block用户进程，而是立刻返回一个error** 。从用户进程角度讲 ，它发起一个read操作后，并不需要等待，而是马上就得到了一个结果。用户进程判断结果是一个error时，它就知道数据还没有准备好，于是它可以再次发送read操作。一旦kernel中的数据准备好了，并且又再次收到了用户进程的system call，那么它马上就将数据拷贝到了用户内存，然后返回。

> nonblocking IO的特点是用户进程需要不断的主动询问kernel数据好了没有。

## I/O 多路复用 ##

IO multiplexing就是我们说的select，poll，epoll，有些地方也称这种IO方式为event driven IO。select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。它的基本原理就是select，poll，epoll这个function会不断的轮询所负责的所有socket，当某个socket有数据到达了，就通知用户进程。

机制：一个线程以阻塞的方式监听客户端请求；另一个线程采用NIO的形式select已经接收到数据的channel信道，处理请求；

*  select，poll，epoll模型 - **处理更多的连接**  
    ![bio2][]  
    上面所说的多路复用的select，poll，epoll本质上都是同步IO，因为他们都需要在读写事件就绪后自己负责进行读写，也就是说这个读写过程是阻塞的，实际上是指阻塞在select上面，必须等到读就绪、写就绪等网络事件。异步IO则无需自己负责进行读写，异步IO的实现会负责把数据从内核拷贝到用户空间。

> I/O 多路复用的特点是通过一种机制一个进程能同时等待多个文件描述符，  
> 而这些文件描述符（套接字描述符）其中的任意一个进入读就绪状态，select()  
> 函数就可以返回。所以，如果处理的连接数不是很高的话，使用select/epoll的web  
> server不一定比使用multi-threading + blocking IO的web  
> server性能更好，可能延迟还更大。select/  
> epoll的优势并不是对于单个连接能处理得更快，而是在于能处理更多的连接。

*  一个面试问题：select、poll、epoll的区别?

### Java中的I/O 多路复用： Reactor模型 ###

(主从Reactor模型）netty就是主从Reactor模型的实现，相当于这个模型在  
![bio3][]

对比与传统的I/O 多路复用，Reactor模型增加了事件分发器，基于事件驱动，能够将相应的读写事件分发给不同的线程执行，真正实现了非阻塞I/O。

基于Reactor Pattern 处理模式中，定义以下三种角色

>  *  Reactor将I/O事件分派给对应的Handler
>  *  Acceptor处理客户端新连接，并分派请求到处理器链中
>  *  Handlers执行非阻塞读/写 任务

举个栗子  
回顾我们上面写的代码，是不是每个线程处理一个连接，显然在高并发情况下是不适用的，应该采用 **IO多路复用** 的思想，使得一个线程能够处理多个连接，并且不能阻塞读写操作，添加一个 **选择器**在buffer有数据的时候就开始写入用户空间.这里的多路是指N个连接，每一个连接对应一个channel，或者说多路就是多个channel。复用，是指多个连接复用了一个线程或者少量线程

现在我们来优化下上面的socket IO模型

优化后的IO模型：

实现一个最简单的Reactor模式：注册所有感兴趣的事件处理器，单线程轮询选择就绪事件，执行事件处理器。流程就是不断轮询可以进行处理的事件，然后交给不同的handler进行处理.  
上面提到的主要是四个网络事件：有连接就绪，接收就绪，读就绪，写就绪。I/O复用主要是通过 **Selector复用器**来实现的，可以结合下面这个图理解上面的叙述

![io mul][]

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

public class NIOServer {
    
        private ServerSocketChannel serverSocket;
        private Selector selector;
        private ReadHandler readHandler;
        private WriteHandler writeHandler;
        private ExecutorService executorService = Executors.newCachedThreadPool();
    
        abstract class Handler {
            protected SelectionKey key;
        }
    
        class ReadHandler extends Handler implements Runnable {
    
            @Override
            public void run() {
                ///...读操作
            }
        }
    
        class WriteHandler extends Handler implements Runnable {
            @Override
            public void run() {
                ///...写操作
            }
        }
    
        public NIOServer(int port) {
            try {
                selector = Selector.open();
                serverSocket = ServerSocketChannel.open();
                serverSocket.bind(new InetSocketAddress(port));
                serverSocket.register(this.selector, SelectionKey.OP_ACCEPT);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    selector.select();  //阻塞等待事件
                    Iterator<SelectionKey> iterator = this.selector.keys().iterator();  // 事件列表 , key -> channel ，每个KEY对应了一个channel
                    while (iterator.hasNext()) {
                        iterator.remove();
                        dispatch(iterator.next());  //分发事件
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
    
            }
        }
    
        private void dispatch(SelectionKey key) {
            if (key.isAcceptable()) {
                register(key);  //新连接建立，注册一个新的读写处理器
            } else if (key.isReadable()) {
                this.executorService.submit(new ReadHandler(key));  //可以写，执行写事件
            } else if (key.isWritable()) {
                this.executorService.submit(new WriteHandler(key));  //可以读。执行读事件
            }
        }
    
        private void register(SelectionKey key) {
            ServerSocketChannel channel = (ServerSocketChannel) key.channel();   //通过key找到对应的channel
            try {
                SocketChannel socketChannel = channel.accept();
                channel.configureBlocking(false);
                channel.register(this.selector, SelectionKey.OP_ACCEPT);
            } catch (IOException e) {
                e.printStackTrace();
            }
    
        }
    }

### 优化线程模型 ###

上述模型还可以继续优化。因为上述模型只是增多个客户端连接的数量，但是在高并发的情况下，

## 参考资料： ##

[老曹眼中的网络编程基础][Link 1]  
[Linux IO模式及 select、poll、epoll详解][Linux IO_ select_poll_epoll]

[IP]: /images/20220413/2e43fe11c1f643409cba3c13dfe57568.png
[TCP]: /images/20220413/9660784cfc9f4d25a1e2a89f6bf35367.png
[bio]: /images/20220413/21d9e689636543b6bfa9a8af4f9cf6a7.png
[bio1]: /images/20220413/ade149e1e56d46a79e235fd997a6f8e8.png
[bio2]: /images/20220413/267aff31fdaa4ac78646e13e7fc3a899.png
[bio3]: /images/20220413/83f58c9351b6411cb6a3114ef7c004a9.png
[io mul]: /images/20220413/516a9c25e3d54ff2b2953f35d3d8d80b.png
[Link 1]: https://mp.weixin.qq.com/s/XXMz5uAFSsPdg38bth2jAA
[Linux IO_ select_poll_epoll]: https://segmentfault.com/a/1190000003063859