首页 > 编程语言 >Java中常见的网络通信模型

Java中常见的网络通信模型

时间：2023-07-28 10:23:46浏览次数：35  
标签：网络通信 Java 模型 用户 FD 内核 IO 线程 数据

image-20230727165438152

目前最近仔学习RocketMQ以及Dubbo还有Spring5框架的底层部分，了解到这些技术的底层都是采用的Netty作为底层的通信的软件，于是便需要详细了解以下网络中的通信的模型以及Netty的通信模型原理。

本篇是通过Redis以及Netty进行网络通信模型的逐渐演化来进行介绍，其中还会夹杂着一些比较重要的概念，例如：零拷贝技术等。通过本篇大家会详细的知道整个网络通信的底层模型，此外还会介绍一些重要的概念比如：同步阻塞，同步非阻塞，多路复用，异步阻塞，异步非阻塞

1. 用户空间与内核空间的介绍：

ubuntu和Centos 都是Linux的发行版，发行版可以看成对linux包了一层壳，任何Linux发行版，其系统内核都是Linux。我们的应用都需要通过Linux内核与硬件交互。

用户的应用，比如redis，mysql等其实是没有办法去执行访问我们操作系统的硬件的，所以我们可以通过发行版的这个壳子去访问内核，再通过内核去访问计算机硬件。image-20230727174802144

计算机硬件包括，如cpu，内存，网卡等等，内核（通过寻址空间）可以操作硬件的，但是内核需要不同设备的驱动，有了这些驱动之后，内核就可以去对计算机硬件去进行 内存管理，文件系统的管理，进程的管理等等。

我们想要用户的应用来访问，计算机就必须要通过对外暴露的一些接口，才能访问到，从而简介的实现对内核的操控，但是内核本身上来说也是一个应用，所以他本身也需要一些内存，cpu等设备资源，用户应用本身也在消耗这些资源，如果不加任何限制，用户去操作随意的去操作我们的资源，就有可能导致一些冲突，甚至有可能导致我们的系统出现无法运行的问题，因此我们需要把用户和内核隔离开

进程的寻址空间划分成两部分：内核空间、用户空间

什么是寻址空间呢？我们的应用程序也好，还是内核空间也好，都是没有办法直接去物理内存的，而是通过分配一些虚拟内存映射到物理内存中，我们的内核和应用程序去访问虚拟内存的时候，就需要一个虚拟地址，这个地址是一个无符号的整数，比如一个32位的操作系统，他的带宽就是32，他的虚拟地址就是2的32次方，也就是说他寻址的范围就是0~2的32次方， 这片寻址空间对应的就是2的32个字节，就是4GB，这个4GB，会有3个GB分给用户空间，会有1GB给内核系统

image-20230727182018473

在linux中，他们权限分成两个等级，0和3，用户空间只能执行受限的命令（Ring3），而且不能直接调用系统资源，必须通过内核提供的接口来访问内核空间可以执行特权命令（Ring0），调用一切系统资源，所以一般情况下，用户的操作是运行在用户空间，而内核运行的数据是在内核空间的，而有的情况下，一个应用程序需要去调用一些特权资源，去调用一些内核空间的操作，所以此时他俩需要在用户态和内核态之间进行切换。

2.网络模型：

2.1 阻塞IO模型：

应用程序想要去读取数据，他是无法直接去读取磁盘数据的，他需要先到内核里边去等待内核操作硬件拿到数据，这个过程就是1，是需要等待的，等到内核从磁盘上把数据加载出来之后，再把这个数据写给用户的缓存区，这个过程是2，如果是阻塞IO，那么整个过程中，用户从发起读请求开始，一直到读取到数据，都是一个阻塞状态。

image-20230727183503397

具体流程如下图：

用户去读取数据时，会去先发起recvform一个命令，去尝试从内核上加载数据，如果内核没有数据，那么用户就会等待，此时内核会去从硬件上读取数据，内核读取数据之后，会把数据拷贝到用户态，并且返回ok，整个过程，都是阻塞等待的，这就是阻塞IO

总结如下：

顾名思义，阻塞IO就是两个阶段都必须阻塞等待：

阶段一：

  • 用户进程尝试读取数据，例如是网卡数据
  • 此时数据尚未到达，内核需要等待相应的数据到达
  • 此时用户进程也处于阻塞的状态

阶段二：

  • 数据到达拷贝到内核缓冲区，代表已经就绪
  • 将内核数据拷贝到用户缓冲区
  • 拷贝的过程当中用户进程依然处于阻塞等待的状态
  • 拷贝完成，用户进程接触阻塞，处理相应的数据

可以看到，在阻塞IO模型当中，用户进程在两个阶段都是阻塞的状态：
image-20230727184008523

2.2. 非阻塞IO模型：

顾名思义，非阻塞IO的recvfrom操作会立即返回结果而不是阻塞用户的进程：

阶段一：

  • 用户进程尝试读取相应的数据，例如网卡数据
  • 此时数据尚未到达，内核需要等待数据
  • 返回异常给用户进程
  • 用户进程拿到error后会再次尝试进行获取
  • 循环往复，直到数据就绪

阶段二：

  • 将内核数据拷贝到用户缓冲区
  • 拷贝过程当中，用户进程依然阻塞等待
  • 拷贝完成后，用户进程解除阻塞，处理相应的数据
  • 可以看到，非阻塞IO模型当中，用户进程在第一阶段是非阻塞，第二个阶段是阻塞状态。虽然是非阻塞，但是性能并没有得到什么提高。而且忙等机制会导致CPU空转。CPU的使用率暴增。

image-20230727184610679

2.3 IO多路复用的方式：

改进的策略就是当用户进程发现内核中数据就绪才来进行处理，在数据未就绪的时候CPU不会空转或者是空等。

如果直到用户进程是否就绪，那么就需要下面的数据结构：

文件描述符（FD）：简称FD，是一个从0 开始的无符号整数，用来关联Linux中的一个文件。在Linux中，一切皆文件，例如常规文件、视频、硬件设备等，当然也包括网络套接字（Socket）。

通过FD，我们的网络模型可以利用一个线程监听多个FD，并在某个FD可读、可写时得到通知，从而避免无效的等待，充分利用CPU资源。

image-20230727194428848

阶段一：

  • 用户进程调用select，指定要监听的FD集合
  • 核监听FD对应的多个socket
  • 任意一个或多个socket数据就绪则返回readable
  • 此过程中用户进程阻塞

阶段二：

  • 用户进程找到就绪的socket
  • 依次调用recvfrom读取数据
  • 内核将数据拷贝到用户空间
  • 用户进程处理数据

当用户去读取数据的时候，不再去直接调用recvfrom了，而是调用select的函数，select函数会将需要监听的数据交给内核，由内核去检查这些数据是否就绪了，如果说这个数据就绪了，就会通知应用程序数据就绪，然后来读取数据，再从内核中把数据拷贝给用户态，完成数据处理，如果N多个FD一个都没处理完，此时就进行等待。

用IO复用模式，可以确保去读数据的时候，数据是一定存在的，他的效率比原来的阻塞IO和非阻塞IO性能都要高

IO多路复用是利用单个线程来同时监听多个FD，并在某个FD可读、可写时得到通知，从而避免无效的等待，充分利用CPU资源。不过监听FD的方式、通知的方式又有多种实现，常见的有：

  • select
  • poll
  • epoll

其中select和poll相当于是当被监听的数据准备好之后，他会把你监听的FD整个数据都发给你，你需要到整个FD中去找，哪些是处理好了的，需要通过遍历的方式，所以性能也并不是那么好

而epoll，则相当于内核准备好了之后，他会把准备好的数据，直接发给你，咱们就省去了遍历的动作。

2.3.1 IO多路复用当中的select方式：

image-20230727194816564

select函数包括如下的数据：
   1.nfds:要监听的fd_set的最大的fd+1
   2.fd_set *readfds:要监听读事件的fd的集合
   3.fd_set *writefds:要监听写事件的fd的集合
   4.fd_set *exceptfds:要监听异常事件的fd集合
   5.struct timeval *timeout:超时时间
fd_set:
   代表的是1024个bit位，每一个bit位代表一个fd 0代表未就绪，1代表就绪

比如要监听的数据，是1,2,5三个数据，此时会执行select函数，然后将整个fd发给内核态，内核态会去遍历用户态传递过来的数据，如果发现这里边都数据都没有就绪，就休眠，直到有数据准备好时，就会被唤醒，唤醒之后，再次遍历一遍，看看谁准备好了，然后再将处理掉没有准备好的数据，最后再将这个FD集合写回到用户态中去，此时用户态就知道了，奥，有人准备好了，但是对于用户态而言，并不知道谁处理好了，所以用户态也需要去进行遍历，然后找到对应准备好数据的节点，再去发起读请求，我们会发现，这种模式下他虽然比阻塞IO和非阻塞IO好，但是依然有些麻烦的事情， 比如说频繁的传递fd集合，频繁的去遍历FD等问

注意在用户态转换成为内核态的时候就会触发select()的调用，此时内核态会遍历fd来发现谁准备好了，如果准备好了就会返回，因为select()有返回值，返回值是int就是告诉用户态到底是谁准备好了数据，但是并不知道到底是哪个fd准备好了数据，因此就需要遍历。

这种模式存在的问题：

  • 需要将整个fd_set从用户空间拷贝到内核空间，select结束以后好需要再次拷贝回用户空间
  • select无法得知具体是哪个fd就绪的，需要遍历整个fd_set
  • fd_set监听的fd数量不能超过1024

2.3.2 IO多路复用中的poll模式：

poll模式对select模式做了简单改进，但性能提升不明显，部分关键代码如下：

IO流程：

  • 创建pollfd数组，向其中添加关注的fd信息，数组大小自定义
  • 调用poll函数，将pollfd数组拷贝到内核空间，转链表存储，无上限
  • 内核遍历fd，判断是否就绪
  • 数据就绪或超时后，拷贝pollfd数组到用户空间，返回就绪fd数量n
  • 用户进程判断n是否大于0,大于0则遍历pollfd数组，找到就绪的fd

与select对比：

  • select模式中的fd_set大小固定为1024，而pollfd在内核中采用链表，理论上无上限
  • 监听FD越多，每次遍历消耗时间也越久，性能反而会下降

image-20230727195903627

2.3.3 IO多路复用中的epoll模式：

epoll模式是对select和poll的改进，它提供了三个函数：

image-20230727200014432

第一个是：eventpoll的函数，他内部包含两个东西

一个是：

1、红黑树-> 记录的事要监听的FD

2、一个是链表->一个链表，记录的是就绪的FD

紧接着调用epoll_ctl操作，将要监听的数据添加到红黑树上去，并且给每个fd设置一个监听函数，这个函数会在fd数据就绪时触发，就是准备好了，现在就把fd把数据添加到list_head中去

3、调用epoll_wait函数

就去等待，在用户态创建一个空的events数组，当就绪之后，我们的回调函数会把数据添加到list_head中去，当调用这个函数的时候，会去检查list_head，当然这个过程需要参考配置的等待时间，可以等一定时间，也可以一直等， 如果在此过程中，检查到了list_head中有数据会将数据添加到链表中，此时将数据放入到events数组中，并且返回对应的操作的数量，用户态的此时收到响应后，从events中拿到对应准备好的数据的节点，再去调用方法去拿数据。

epoll模式中如何解决这些问题的？

  • 基于epoll实例中的红黑树保存要监听的FD，理论上无上限，而且增删改查效率都非常高
  • 每个FD只需要执行一次epoll_ctl添加到红黑树，以后每次epol_wait无需传递任何参数，无需重复拷贝FD到内核空间
  • 利用ep_poll_callback机制来监听FD状态，无需遍历所有FD，因此性能不会随监听的FD数量增多而下降

补充：

网络模型epoll中的ET和LT：

就是IO多路复用当中的事件通知机制：

当FD有数据可读时，我们调用epoll_wait（或者select、poll）可以得到通知。但是事件通知的模式有两种：

  • LevelTriggered：简称LT，也叫做水平触发。只要某个FD中有数据可读，每次调用epoll_wait都会得到通知。
  • EdgeTriggered：简称ET，也叫做边沿触发。只有在某个FD有状态变化时，调用epoll_wait才会被通

举个例子：

image-20230727201251492

结论

如果我们采用LT模式，因为FD中仍有1kb数据，则第⑤步依然会返回结果，并且得到通知
如果我们采用ET模式，因为第③步已经消费了FD可读事件，第⑤步FD状态没有变化，因此epoll_wait不会返回，数据无法读取，客户端响应超时。

那么有人就会问了ET模式下剩余数据如何处理，因此我们需要手动进行处理再次调用epoll_ctl函数去判断红黑树中是否有剩余的已经准备好的数据是否没有读完毕，如果存在就会重新加入到list_head链表当中。

LT缺点是会出现惊群现象，就是可能存在多个等待list_head中有数据的进程，此时如果采用LT模式，如果真的list_head中有数据了有可能会导致一部分进程能获取到相对应的数据，当消费完毕list_head中的数据以后，可能还会剩余一些进程虽然等待list_head但迟迟等不到数据，就会浪费进程的资源，这就是惊群现象，但是ET模式就没有上述的问题。

2.3.4 基于epoll的服务器端的流程：

image-20230727201442077

服务器启动以后，服务端会去调用epoll_create，创建一个epoll实例，epoll实例中包含两个数据

1、红黑树（为空）：rb_root 用来去记录需要被监听的FD

2、链表（为空）：list_head，用来存放已经就绪的FD

第二步：创建好了之后，会去调用epoll_ctl函数，此函数会会将需要监听的数据添加到rb_root中去，并且对当前这些存在于红黑树的节点设置回调函数，当这些被监听的数据一旦准备完成，就会被调用，而调用的结果就是将红黑树的fd添加到list_head中去(但是此时并没有完成)

第三步：当第二步完成后，就会调用epoll_wait函数，这个函数会去校验是否有数据准备完毕（因为数据一旦准备就绪，就会被回调函数添加到list_head中），在等待了一段时间后(可以进行配置)，如果等够了超时时间，则返回没有数据，如果有，则进一步判断当前是什么事件，如果是建立连接时间，则调用accept() 接受客户端socket，拿到建立连接的socket，然后建立起来连接，如果是其他事件，则把数据进行写出

3. Redis的网络模型：

3.1 Redis是单线程的嘛：

我们经常会在面试的时候被问到redis是单线程还是多线程模型，如果回答单线程，那么面试官又会问为什么使用单线程模型：

Redis到底是单线程还是多线程？

  • 如果仅仅聊Redis的核心业务部分（命令处理），答案是单线程
  • 如果是聊整个Redis，那么答案就是多线程

为什么redis要选择单线程？

  • 抛开持久化不谈，Redis是纯 内存操作，执行速度非常快，它的性能瓶颈是网络延迟而不是执行速度，因此多线程并不会带来巨大的性能提升。
  • 多线程会导致过多的上下文切换，带来不必要的开销
  • 引入多线程会面临线程安全问题，必然要引入线程锁这样的安全手段，实现复杂度增高，而且性能也会大打折扣

3.2 Redis单线程模型与多线程模型的网络变更是什么样的：

Redis通过IO多路复用技术来提高网络性能，并且支持各种不同的多路复用的实现，并且将这些实现进行了封装，提供了统一的高性能事件库API库AE：

image-20230727203401483

不同的ae文件对应不同的操作系统。注意redis底层源码当中有一个ae.c的文件会根据不同的操作系统进行判断并赋值相应的变量。

注意对于不同的操作系统，redis为封装了基本是统一的方法接口：

image-20230727203753576

接下来来看一下redis单线程网络模型当中的整个流程：

1.整个redis启动的入口在server.c的文件当中，也就是整个redis启动的主函数：

image-20230727204300088

2.紧接着在初始化服务当中我们就会发现进行了epoll_create的创建（aeCreateEventLoop函数）就是创建了红黑树与链表，其实就是创建ServerSocket（listenToPort）,同时监听端口，此时的server.port就是调用的redis的默认配置的端口6379，并且配置主机的访问IP地址，得到FD，并且调用createSocketAcceptHandler内部会调用aeApiAddEvent监听FD，如果此时有客户端进行访问就会调用acceptTcpHandler进行处理：

image-20230727204407337

注意在完成监听FD工作后，在调用epoll_wait之前我们调用了aeSetBeforeSleepProc函数做了一些前置的处理工作。

3.再次查看main函数发现在initServer()完毕后才开是进行监听事件的循环aeMain(server.el):

注意里面的aeApiPoll函数返回的numevents就是就绪的FD的数量，当numevents大于0的时候说明就是有FD就绪了，就可以进行处理了，注意此时调用的处理的函数就是acceptTcpHandler函数：

image-20230727205946596

因此接下来查看acceptTcpHandler:

注意一开始就是我们客户端需要来进行连接的FD所对应的acceptTcpHandler:

image-20230727210424201

如果是客户端Socket可读那么就是readQueryFromClient进行处理：image-20230727210510813

那么Redis的底层网络模型详解的流程图如下：
image-20230727210725647

注意在处理客户端客户端可读的时候就是readable事件时候会触发readQueryFromClient函数此时会将客户端的数据请求放置到c->queryBuf缓冲区当中，随后会解析客户端的redis命令将结果放到buf或者是reply当中。此时注意redis底层会有一个listAddNodeHead函数，会将客户端的解析后的结果均存在server.clients_pending_write队列当中。注意当客户端源源不断地请求过来时候就会有很多解析redis后的命令被放在这个队列当中。接下来就需要分析我们前面一直没有分析过的beforesleep函数了：

image-20230727212541819

  • 首先会定义一个迭代器指向server.clients_pending_write的队列的头部
  • 迭代的方向是AL_START_HEAD从头向后进行迭代
  • 再往下我们就会发现了writeHandler函数，从中会有sendReplyToClient函数监听的是客户端writeable事件一个卸到客户端的Socket当中

注意在redis6.0以后会引入多线程：

redis6.0版本中引入了多线程，目的就是提高IO读写效率的问题。因此在解析客户端命令，写响应结果时候采用了多线程。核心的命令执行，IO多路复用依然是由主线程执行。

4.几种不同的IO的模型的整理：

我们常见的IO模型有

  • 同步阻塞IO（BIO），
  • 同步非阻塞IO（NIO），
  • IO多路复用：即经典的Reactor设计模式，有时候也被称之为异步阻塞IO。
  • 异步IO（AIO),有时候也被称之为异步非阻塞的IO。

5. Reactor模型：

Reactor线程模型是基于NIO的，有三种模式，分别介绍如下：

  • 单Reator单线程模型;

image-20230727214713978

Reactor可以理解为主线程（或者单独创建一个线程去运行下述的代码，此线程就叫作Reactor，主线程则用来阻塞不退出），运行的逻辑就是通过select监控客户端请求的事件，并且对于一些连接进行read->业务处理->send，下面的代码就是此模式：
image-20230727215152259

  • 单Reactor-多线程：

image-20230727215346846

  1. Reactor对象通过select监控客户端请求的事件，收到请求以后，通过Dispatch进行分发
  2. 如果建立连接请求，则由Acceptor通过accept处理连接请求，然后创建一个Handler对象处理完成连接后的各种事件
  3. 如果不是连接请求。则由Reactor分发调用连接对应的handler来进行处理
  4. handler只是负责响应事件，不做具体的业务处理，通过read读取数据后，会分发给后面的worker线程池的某一个线程进行处理
  5. worker线程池会独立的分配相应独立的线程完成真正的任务，并将结果返回给handler
  6. handler收到响应以后，通过send将结果返回给client

我们在这里给出一个NIO多线程的版本，与Netty底层通信原理很像，我们一步步演化：

第一步：首先搭建好worker架构并且将读事件与worker关联：

package cn.itcast.netty.c2;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicInteger;

import static cn.itcast.netty.c1.ByteBufferUtil.debugAll;

//采用多线程优化NIO：
@Slf4j
public class MultiThreadServer {
    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        Selector boss = Selector.open();
        SelectionKey bossKey = ssc.register(boss, 0, null);
        bossKey.interestOps(SelectionKey.OP_ACCEPT);
        ssc.bind(new InetSocketAddress(9000));
        //这里我们假设只使用一个worker，之后我们还会优化使用多个worker:
        Worker worker = new Worker("worder-0");
        worker.register();//创建线程和selector

        while(true){
            boss.select();
            Iterator<SelectionKey> iter = boss.selectedKeys().iterator();
            while(iter.hasNext()){
                SelectionKey key = iter.next();
                iter.remove();
                if(key.isAcceptable()) {
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);
                    log.debug("connected.....{}",sc.getRemoteAddress());
      
                    log.debug("before register.....{}",sc.getRemoteAddress());
                    
                    //关联selector:
                    sc.register(worker.selector,SelectionKey.OP_READ,null);
                    
                    log.debug("after register.....{}",sc.getRemoteAddress());
                }
            }
        }
    }
    //worker主要处理读写事件：
    static class Worker implements Runnable{
        private Thread thread;
        private Selector selector;
        private String name;
        
        //保证线程的创建与selector只会执行一次不会多次执行：
        private volatile boolean start = false;

        public Worker(String name)
        {
            this.name = name;
        }
        //初始化线程与selector:
        public void register() throws IOException{
            if(!start){
                //this就代表当前worder对象需要执行的代码，也就是run()中需要执行的代码：
                thread = new Thread(this,name);
                thread.start();
                selector = Selector.open();
                start = true;
            }
        }

        //run()方法其实就是检测读写事件：
        @Override
        public void run() {
            while(true){
                try {
                    selector.select();
                    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                    while(iter.hasNext()){
                        SelectionKey key = iter.next();
                        iter.remove();
                        if(key.isReadable()){
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            SocketChannel channel = (SocketChannel) key.channel();
                            log.debug("read...{}",channel.getRemoteAddress());
                            channel.read(buffer);
                            buffer.flip();
                            debugAll(buffer);
                        }
                    }
                }catch (IOException e){
                    e.printStackTrace();
                }
            }
        }
    }
}

上述代码会导致客户端发送的数据无法正确的被服务器端接收，原因如下：
我们需要重新的审核一下整个代码的流程，注意如下的两个代码片段：
image-20230728084741809

image-20230728084801296

注意第一个代码片段中的sc.register代码是在boss线程中执行的，第二个代码片段是在worker线程中执行的，他们共同使用的是同一个selector。那么就会造成两个不同的现象：

1.如果代码2片段比代码1片段先要执行，就会在客户端未到来时候在select()阻塞运行，导致register方法也不会向下运行，只有当客户端线程进行连接后，才会导致select方法向下执行，同时使得register向下执行。这就会导致客户端写的数据服务器端始终收不到

2.但是如果两个代码片段反过来的顺序执行，那么服务器端便会正常的收取到客户端发送过来的数据。

第二步：将sc.register与selector.select()放置在同一线程中运行，这样能够控制程序的运行顺序：

image-20230728090332945

这里我们采用一个队列进行处理：

image-20230728090924115

注意此时我们还需要唤醒一下seletor防止阻塞：

image-20230728091245685

我们来理一下顺序：

1.首先客户端建立连接后会调用work.register(sc)进行初始化。
2.如果是第一次执行就会创建selector与thread,并且work-0线程也就会启动。
3.紧接着我们会向消息队列中存放一个注册读消息的事件
4.此时work-0因为会被thread.start()会被执行，然后在select()方法阻塞住，此时我们采用在boss线程中唤醒selector.wakeup()；
5.在唤醒后会注册相应的读事件，此时可能还未有读事件，于是又到达while循环的select()方法阻塞住
6.当再有事件来的时候select()解除就会读取客户端发来的数据了。

第三步：之前两步我们都是基于work-0一个线程做的操作，接下来我们采用多线程。

package cn.itcast.netty.c2;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicInteger;

import static cn.itcast.netty.c1.ByteBufferUtil.debugAll;

//采用多线程优化NIO：
@Slf4j
public class MultiThreadServer {
    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        Selector boss = Selector.open();
        SelectionKey bossKey = ssc.register(boss, 0, null);
        bossKey.interestOps(SelectionKey.OP_ACCEPT);
        ssc.bind(new InetSocketAddress(9000));
        //这里我们假设只使用一个worker，之后我们还会优化使用多个worker:
        //Worker worker = new Worker("worder-0");
        //worker.register();//创建线程和selector
        //充分发挥多核cpu的性能，可以将cpu的核数作为worker的长度：
        Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];//注意在Docker中会有Bug
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new Worker("worker-" + i);
        }
        //定义一个计数器：默认初始是0，然后依次自增
        AtomicInteger index = new AtomicInteger();

        while(true){
            boss.select();
            Iterator<SelectionKey> iter = boss.selectedKeys().iterator();
            while(iter.hasNext()){
                SelectionKey key = iter.next();
                iter.remove();
                if(key.isAcceptable()) {
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);
                    log.debug("connected.....{}",sc.getRemoteAddress());
                    //注意new Worker不应该放在此位置，因为如果由10个连接就会创建10个worker,worker不应该由连接数决定
                    //关联worker:
                    log.debug("before register.....{}",sc.getRemoteAddress());

                    //使用轮询算法：
                    workers[index.getAndIncrement() % workers.length].register(sc);

                    //sc.register(worker.selector,SelectionKey.OP_READ,null);
                    log.debug("after register.....{}",sc.getRemoteAddress());
                }
            }
        }
    }
    //worder主要处理读写事件：
    static class Worker implements Runnable{
        private Thread thread;
        private Selector selector;
        private String name;
        //两个线程之间传递数据：消息队列进行解耦：
        private ConcurrentLinkedDeque<Runnable> queue = new ConcurrentLinkedDeque<>();

        //保证线程的创建与selector只会执行一次不会多次执行：
        private volatile boolean start = false;

        public Worker(String name)
        {
            this.name = name;
        }
        //初始化线程与selector:
        public void register(SocketChannel sc) throws IOException{
            if(!start){
                //this就代表当前worder对象需要执行的代码，也就是run()中需要执行的代码：
                thread = new Thread(this,name);
                thread.start();
                selector = Selector.open();
                start = true;
            }
            //使用消息队列解耦，将数据传递到work-0线程当中：
            queue.add(()->{
                try {
                    sc.register(selector,SelectionKey.OP_READ,null);
                } catch (ClosedChannelException e) {
                    e.printStackTrace();
                }
            });
            selector.wakeup();//唤醒selector
        }

        //run()方法其实就是检测读写事件：
        @Override
        public void run() {
            while(true){
                try {
                    selector.select();
                    Runnable task = queue.poll();
                    if(task != null)
                    {
                        task.run();//注意这里不会创建新的线程，因为run方法需要配合start()才会创建新的线程
                    }
                    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                    while(iter.hasNext()){
                        SelectionKey key = iter.next();
                        iter.remove();
                        if(key.isReadable()){
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            SocketChannel channel = (SocketChannel) key.channel();
                            log.debug("read...{}",channel.getRemoteAddress());
                            channel.read(buffer);
                            buffer.flip();
                            debugAll(buffer);
                        }
                    }
                }catch (IOException e){
                    e.printStackTrace();
                }
            }
        }
    }
}
  • 主从Reactor与多线程

image-20230727220336428

  1. reactor主线程MainReactor对象通过select监听连接事件，收到事件以后，通过Acceptor处理连接事件
  2. 当Acceptor处理连接事件后，MianReactor将连接分配给SubReactor
  3. Subreactor将连接加入到连接队列进行监听，并创建Handler进行各种事件的处理
  4. 当有新的事件发生的时候，subReactor就会调用对应的handler进行处理
  5. handler通过read读取数据，分发给后面的worker线程进行处理
  6. worker线程池分配独立的worker线程进行业务的处理，并返回结果
  7. handler收到响应的结果以后，再通过send将结果返回给client
  8. Reactor主线程可以对应多个reactor子线程，即MainReactor可以关联多个SubReactor。

6. NIO中的零拷贝技术：

1.首先来看一下传统文件的传输：

如下图所示是我们使用的IO流读取磁盘文件，通过SocketAPI发送的流程，其中需要read,write系统低调用，每一次调用都意味着上下文的切换过程，即用户态与内核态的上下文的切换的过程，并且这一过程当中存在四次的数据拷贝，其中两次由DMA负责打工，两次由CPU负责拷贝：
image-20230728094853485

优化的策略：

  • 如果java程序不需要对于磁盘的内容数据进行再次的加工，那么就不需要拷贝到用户空间，从而减少相应的拷贝的次数。
  • 由于用户空间没有操作网卡和磁盘的权限，那么操作这些设备就需要操作系统内核来完成，那么如果操作系统提供新的系统调用函数，岂不是就可以减少用户态与内核态的上下文的切换的过程。

优化的策略如下：

1.采用mmap+write操作：

image-20230728095246916

  • 进程调用mmap函数以后，DMA会把磁盘的数据拷贝到内核的缓冲区内部。紧接着，应用进程跟操作系统内核共享这个缓冲区。
  • 应用进程再次调用write函数后，操作系统直接将内核缓冲区的数据拷贝到socket缓冲区当中吗，这一切都发生在内核态，由CPU来搬运相应的数据。
  • 最后把内核的socket缓冲区里面的数据拷贝到网卡的缓冲区当中，这个过程是由DMA搬运的。

mmap优化的步骤在于并没有减少系统调用带来的内核态与用户态之间的切换，只是应用程序和内核共享了缓冲区，从而让cpu可以直接将内核缓冲区的数据拷贝到socket缓冲区，不需要拷贝到用户缓冲区，再从用户缓冲区拷贝到socket缓冲区。

2.采用sendFile:

image-20230728100702627

sendFile可以减少write,read导致的系统的调用，从而优化效率。如果网卡支持SG-DMA技术，那么还可以进一步优化：

image-20230728100818575

1.通过DMA将磁盘的数据拷贝到内核缓冲区当中

2.缓冲区描述符和数据长度传到socket缓冲区中，这样网卡的SG-DMA控制器就可以直接将内核缓冲区中的数据拷贝到网卡的缓冲区中，此过程不需要将数据从操作系统内核缓冲区拷贝到socket缓冲区中，这样就减少了一次数据的拷贝工作。

这就是我们所说的零拷贝的技术。

对于以上说的两种技术，java在api层面做了相应的封装：

1.FileChannel中的map函数

image-20230728101219502

如上是MappedByteBuffer的获取方式，，其底层是通过反射的技术调用的DirectByteBuffer的构造方法实现的。

2.FileChannel中的transferTo与transFrom函数是对于sendFile函数的封装：

image-20230728101423464

在操作系统层面是调用的一个sendFile系统调用。通过这个系统调用，可以在内核层直接完成文件内容的拷贝。

标签：网络通信,Java,模型,用户,FD,内核,IO,线程,数据
From： https://www.cnblogs.com/liuyuchengcj/p/17586879.html

相关文章

  • Java复制
    将一个对象的引用复制给另外一个对象，一共有三种方式。第一种是直接赋值，第二种方式是浅拷贝，第三种是深拷贝。所以大家知道了哈，这三种概念都是为了拷贝对象。直接赋值复制直接赋值。在Java中，Aa1=a2，我们需要理解的是这实际上复制的是引用，也就是说a1和a2指向的是同一个对象。因......
  • java项目更改jdk版本出现问题
    这里demo中的java版本出了点问题，将jdk17改为jdk11打开项目结构设置Project中的SDK和Languagelevel为jdk11。一、java：错误：不支持发行版本17 打开settings中的JavaCompiler修改module中项目的java版本二、java:错误:无效的源发行版：17 打开项目结构，找到下图界面修改J......
  • java面试题带链接
    下面是在网下下载的，不知道哪来的了，发出来方便查看一、Java基础1.String类为什么是final的。https://www.zhihu.com/question/313455922.HashMap的源码，实现原理，底层结构。get(key)方法时获取key的hash值，计算hash&(n-1)得到在链表数组中的位置first=tab[hash&(n-1)],先判断first的k......
  • 要实现Java中的Excel导入导出功能，可以使用Apache POI库。
    下面是一个简单的示例：导入Excel文件：importorg.apache.poi.ss.usermodel.*;importorg.apache.poi.xssf.usermodel.XSSFWorkbook;importjava.io.FileInputStream;importjava.io.IOException;publicclassExcelImportExample{publicstaticvoidmain(String[]args){......
  • OSI 的七层模型都有哪些？
    OSI（OpenSystemsInterconnection）是一个开放系统互联参考模型，用于规范计算机网络体系结构和通信协议。OSI模型定义了七个不同的层，每个层负责特定的功能，从物理连接到应用层协议，实现了网络通信的抽象和分层。以下是OSI模型的七层：物理层（PhysicalLayer）：负责传输比特流（0和1）通过物理媒介（......
  • 在Odoo中，res.partner和res.users模型的关联和区别。
    res.partner模型：这个模型在Odoo中代表一个业务伙伴。业务伙伴可以是一个公司或者是一个个人。                       这个模型用于存储业务伙伴的信息，比如名称、地址、联系方式等。res.partner模型也被用于各种不同的场景，比......
  • Java 大后端各种架构图汇总
    1.java类加载器架构2.JVM架构3.Java技术体系4.线程运行架构5.Java体系（编译与运行）结构6.JMS技术架构7.JMX技术架构8.Spring架构9.Hibernate架构10.ibatis架构11.Struts2架构12.Struts1架构13.JBPM14.EJB技术架......
  • SHA1签名算法，JAVA和C#
    java:publicstaticvoidmain(String[]args)throwsNoSuchAlgorithmException{Stringtoken="31a4a1aa-cffc-4aca-9ef6-0497edf7fbed";Stringnonce="Rzem0rlz19e6GZuZuFKyDzaxiS4baaqn8uvxVnntXKS";Stringtimestamp="1646......
  • JavaAPI
    Scanner类该类是文本扫描类构造方法//创建一个扫描输入内容的扫描器Scannersc=newScanner(Sytem.in);//扫描文本内容的扫描器Filefile=newFile("C:\\Users\\Administrator\\Desktop\\test.txt");Scannersc=newScanner(file);类中方法//从键盘中读取内容int......
  • JavaScript中this关键字详解（二）
    使用注意点避免多层this由于this的指向是不确定的，所以切勿在函数中包含多层的this。varo={f1:function(){console.log(this);//objectvarf2=function(){console.log(this);//window}();}}​o.f1()//Object//Window上面代码......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99