lxKLj

lxKLj 查看完整档案

上海编辑中南林业科技大学  |  软件工程 编辑闪马智能  |  Java开发 编辑 github.com/lxKLj 编辑
编辑

学无止境,笨鸟先飞

个人动态

lxKLj 发布了文章 · 3月3日

ThreadLocal 原理解析

ThreadLocal 核心方法 set 和 get

ThreadLocal 核心对外方法就是set 和 get,通过set 存值,get 取值。

/**
 * 存值
 */
public void set(T value) {
    // 获取当前线程
    Thread t = Thread.currentThread();
    // 获取当前线程中的ThreadLocalMap
    ThreadLocalMap map = getMap(t);
    if (map != null)
        // ThreadLocalMap 已经实例化,直接存值
        map.set(this, value);
    else
        // 实例化ThreadLocalMap
        createMap(t, value);
}

/**
 * 取值
 */
public T get() {
    // 获取当前线程
    Thread t = Thread.currentThread();
    // 获取当前线程中的ThreadLocalMap
    ThreadLocalMap map = getMap(t);
    if (map != null) {
        // ThreadLocalMap 已经实例化,直接取值
        ThreadLocalMap.Entry e = map.getEntry(this);
        if (e != null) {
            @SuppressWarnings("unchecked")
            // 类型强转
            T result = (T)e.value;
            return result;
        }
    }
    // 实例化ThreadLocalMap,并返回初始化值(ThreadLocal 默认null)
    return setInitialValue();
}

从上面两个方法不难看出,ThreadLocal 存取值得空间就是ThreadLocalMap,而通过追踪getMap 我们可以发现,该属性是Thread 类的成员变量,也就是说每个线程都会自带ThreadLocalMap,这也就是各线程ThreadLocal 中的值能够线程隔离的根本原因。

/* ThreadLocal values pertaining to this thread. This map is maintained
 * by the ThreadLocal class. */
ThreadLocal.ThreadLocalMap threadLocals = null;

ThreadLocal 中的set 和get 方法在调用时,如果发现ThreadLocalMap 是空,都会进行实例化操作,代码如下所示

/**
 * 创建ThreadLocalMap
 */
void createMap(Thread t, T firstValue) {
    // 构造函数实例化ThreadLocalMap
    t.threadLocals = new ThreadLocalMap(this, firstValue);
}

/**
 * table 初始大小
 */
private static final int INITIAL_CAPACITY = 16;

/**
 * ThreadLocalMap 构造函数
 */
ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
    // 核心为Entry 类型的数组
    table = new Entry[INITIAL_CAPACITY];
    // 计算存储的数组位置,同HashMap 相似,通过位运算加快计算速度
    int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
    // 构造Entry 对象存入数组中
    table[i] = new Entry(firstKey, firstValue);
    size = 1;
    setThreshold(INITIAL_CAPACITY);
}

ThreadLocalMap 实例化完成后,就是往里面存值的操作,通过构造Entry 对象来存入ThreadLocalMap 中的table 数组中。

private void set(ThreadLocal<?> key, Object value) {

    Entry[] tab = table;
    int len = tab.length;
    // 计算数组下标
    int i = key.threadLocalHashCode & (len-1);
    
    // 遍历数组
    for (Entry e = tab[i];
         e != null;
         e = tab[i = nextIndex(i, len)]) {
        // tab[i] 有Entry 对象
        ThreadLocal<?> k = e.get();
        // 相同的ThreadLocal 对象,值覆盖
        if (k == key) {
            e.value = value;
            return;
        }
        // 原ThreadLocal 对象已经被回收,但由于Entry 是弱引用
        // Entry 的key会被置为null,但是value 依然还在
        // 这也是ThreadLocal 可能会导致内存泄露的原因
        if (k == null) {
            // 替换因ThreadLocal已被回收而废弃的Entry
            replaceStaleEntry(key, value, i);
            return;
        }
    }
    // tab[i] 没有对象
    // 创建Entry 插入
    tab[i] = new Entry(key, value);
    int sz = ++size;
    // 扩容操作
    if (!cleanSomeSlots(i, sz) && sz >= threshold)
        rehash();
}

理解set 操作之后,对于get 操作则很容易理解。如下所示,先计算出数组下标后,直接从table 数组中取值即可。

private Entry getEntry(ThreadLocal<?> key) {
    // 计算数组下标
    int i = key.threadLocalHashCode & (table.length - 1);
    // 根据数组下标取值
    Entry e = table[i];
    if (e != null && e.get() == key)
        return e;
    else
        return getEntryAfterMiss(key, i, e);
}

总结

通读ThreadLocal 的set 和get 方法后,我们就会发现,其实整个主要就涉及Thread 和ThreadLocal 两个大类。而ThreadLocal 中存取值的容器ThreadLocalMap 却是Thread 的成员变量,是线程唯一的,这就确保了每个Thread 线程都有自己的ThreadLocalMap 容器,相互之间是独立的,这就是ThreadLocal 存储数据线程隔离的根本原因。但是ThreadLocal 对象并不是线程唯一的,我们可以实例化多个ThreadLocal 对象来进行不同值的存储,但是要记得在用完后进行remove 操作。因为ThreadLocalMap 作为Thread 的成员变量,其生命周期是和Thread 相同的,如果Thread 线程没有回收,但是ThreadLocal 却回收了,就会导致ThreadLocalMap 中会存在很多key 为null 的Entry 对象,并且其中存储的值也不会回收,如果而ThreadLocal 对象多的话,可能会导致内存泄漏。特别是当Thread 是在线程中的时候,Thread 不会进行回收操作,ThreadLocal 不进行remove 而导致内存泄露的概率更加的大。

查看原文

赞 0 收藏 0 评论 0

lxKLj 收藏了文章 · 2月26日

Linux IO模式及 select、poll、epoll详解

注:本文是对众多博客的学习和总结,可能存在理解错误。请带着怀疑的眼光,同时如果有错误希望能指出。

同步IO和异步IO,阻塞IO和非阻塞IO分别是什么,到底有什么区别?不同的人在不同的上下文下给出的答案是不同的。所以先限定一下本文的上下文。

本文讨论的背景是Linux环境下的network IO。

一 概念说明

在进行解释之前,首先要说明几个概念:
- 用户空间和内核空间
- 进程切换
- 进程的阻塞
- 文件描述符
- 缓存 I/O

用户空间与内核空间

现在操作系统都是采用虚拟存储器,那么对32位操作系统而言,它的寻址空间(虚拟存储空间)为4G(2的32次方)。操作系统的核心是内核,独立于普通的应用程序,可以访问受保护的内存空间,也有访问底层硬件设备的所有权限。为了保证用户进程不能直接操作内核(kernel),保证内核的安全,操心系统将虚拟空间划分为两部分,一部分为内核空间,一部分为用户空间。针对linux操作系统而言,将最高的1G字节(从虚拟地址0xC0000000到0xFFFFFFFF),供内核使用,称为内核空间,而将较低的3G字节(从虚拟地址0x00000000到0xBFFFFFFF),供各个进程使用,称为用户空间。

进程切换

为了控制进程的执行,内核必须有能力挂起正在CPU上运行的进程,并恢复以前挂起的某个进程的执行。这种行为被称为进程切换。因此可以说,任何进程都是在操作系统内核的支持下运行的,是与内核紧密相关的。

从一个进程的运行转到另一个进程上运行,这个过程中经过下面这些变化:
1. 保存处理机上下文,包括程序计数器和其他寄存器。
2. 更新PCB信息。
3. 把进程的PCB移入相应的队列,如就绪、在某事件阻塞等队列。
4. 选择另一个进程执行,并更新其PCB。
5. 更新内存管理的数据结构。
6. 恢复处理机上下文。

注:总而言之就是很耗资源,具体的可以参考这篇文章:进程切换

进程的阻塞

正在执行的进程,由于期待的某些事件未发生,如请求系统资源失败、等待某种操作的完成、新数据尚未到达或无新工作做等,则由系统自动执行阻塞原语(Block),使自己由运行状态变为阻塞状态。可见,进程的阻塞是进程自身的一种主动行为,也因此只有处于运行态的进程(获得CPU),才可能将其转为阻塞状态。当进程进入阻塞状态,是不占用CPU资源的

文件描述符fd

文件描述符(File descriptor)是计算机科学中的一个术语,是一个用于表述指向文件的引用的抽象化概念。

文件描述符在形式上是一个非负整数。实际上,它是一个索引值,指向内核为每一个进程所维护的该进程打开文件的记录表。当程序打开一个现有文件或者创建一个新文件时,内核向进程返回一个文件描述符。在程序设计中,一些涉及底层的程序编写往往会围绕着文件描述符展开。但是文件描述符这一概念往往只适用于UNIX、Linux这样的操作系统。

缓存 I/O

缓存 I/O 又被称作标准 I/O,大多数文件系统的默认 I/O 操作都是缓存 I/O。在 Linux 的缓存 I/O 机制中,操作系统会将 I/O 的数据缓存在文件系统的页缓存( page cache )中,也就是说,数据会先被拷贝到操作系统内核的缓冲区中,然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。

缓存 I/O 的缺点:
数据在传输过程中需要在应用程序地址空间和内核进行多次数据拷贝操作,这些数据拷贝操作所带来的 CPU 以及内存开销是非常大的。

二 IO模式

刚才说了,对于一次IO访问(以read举例),数据会先被拷贝到操作系统内核的缓冲区中,然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。所以说,当一个read操作发生时,它会经历两个阶段:
1. 等待数据准备 (Waiting for the data to be ready)
2. 将数据从内核拷贝到进程中 (Copying the data from the kernel to the process)

正式因为这两个阶段,linux系统产生了下面五种网络模式的方案。
- 阻塞 I/O(blocking IO)
- 非阻塞 I/O(nonblocking IO)
- I/O 多路复用( IO multiplexing)
- 信号驱动 I/O( signal driven IO)
- 异步 I/O(asynchronous IO)

注:由于signal driven IO在实际中并不常用,所以我这只提及剩下的四种IO Model。

阻塞 I/O(blocking IO)

在linux中,默认情况下所有的socket都是blocking,一个典型的读操作流程大概是这样:
clipboard.png

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

所以,blocking IO的特点就是在IO执行的两个阶段都被block了。

非阻塞 I/O(nonblocking IO)

linux下,可以通过设置socket使其变为non-blocking。当对一个non-blocking socket执行读操作时,流程是这个样子:
clipboard.png

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

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

I/O 多路复用( IO multiplexing)

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

clipboard.png

当用户进程调用了select,那么整个进程会被block,而同时,kernel会“监视”所有select负责的socket,当任何一个socket中的数据准备好了,select就会返回。这个时候用户进程再调用read操作,将数据从kernel拷贝到用户进程。

所以,I/O 多路复用的特点是通过一种机制一个进程能同时等待多个文件描述符,而这些文件描述符(套接字描述符)其中的任意一个进入读就绪状态,select()函数就可以返回。

这个图和blocking IO的图其实并没有太大的不同,事实上,还更差一些。因为这里需要使用两个system call (select 和 recvfrom),而blocking IO只调用了一个system call (recvfrom)。但是,用select的优势在于它可以同时处理多个connection。

所以,如果处理的连接数不是很高的话,使用select/epoll的web server不一定比使用multi-threading + blocking IO的web server性能更好,可能延迟还更大。select/epoll的优势并不是对于单个连接能处理得更快,而是在于能处理更多的连接。)

在IO multiplexing Model中,实际中,对于每一个socket,一般都设置成为non-blocking,但是,如上图所示,整个用户的process其实是一直被block的。只不过process是被select这个函数block,而不是被socket IO给block。

异步 I/O(asynchronous IO)

inux下的asynchronous IO其实用得很少。先看一下它的流程:
clipboard.png

用户进程发起read操作之后,立刻就可以开始去做其它的事。而另一方面,从kernel的角度,当它受到一个asynchronous read之后,首先它会立刻返回,所以不会对用户进程产生任何block。然后,kernel会等待数据准备完成,然后将数据拷贝到用户内存,当这一切都完成之后,kernel会给用户进程发送一个signal,告诉它read操作完成了。

总结

blocking和non-blocking的区别

调用blocking IO会一直block住对应的进程直到操作完成,而non-blocking IO在kernel还准备数据的情况下会立刻返回。

synchronous IO和asynchronous IO的区别

在说明synchronous IO和asynchronous IO的区别之前,需要先给出两者的定义。POSIX的定义是这样子的:
- A synchronous I/O operation causes the requesting process to be blocked until that I/O operation completes;
- An asynchronous I/O operation does not cause the requesting process to be blocked;

两者的区别就在于synchronous IO做”IO operation”的时候会将process阻塞。按照这个定义,之前所述的blocking IO,non-blocking IO,IO multiplexing都属于synchronous IO。

有人会说,non-blocking IO并没有被block啊。这里有个非常“狡猾”的地方,定义中所指的”IO operation”是指真实的IO操作,就是例子中的recvfrom这个system call。non-blocking IO在执行recvfrom这个system call的时候,如果kernel的数据没有准备好,这时候不会block进程。但是,当kernel中数据准备好的时候,recvfrom会将数据从kernel拷贝到用户内存中,这个时候进程是被block了,在这段时间内,进程是被block的。

而asynchronous IO则不一样,当进程发起IO 操作之后,就直接返回再也不理睬了,直到kernel发送一个信号,告诉进程说IO完成。在这整个过程中,进程完全没有被block。

各个IO Model的比较如图所示:
clipboard.png

通过上面的图片,可以发现non-blocking IO和asynchronous IO的区别还是很明显的。在non-blocking IO中,虽然进程大部分时间都不会被block,但是它仍然要求进程去主动的check,并且当数据准备完成以后,也需要进程主动的再次调用recvfrom来将数据拷贝到用户内存。而asynchronous IO则完全不同。它就像是用户进程将整个IO操作交给了他人(kernel)完成,然后他人做完后发信号通知。在此期间,用户进程不需要去检查IO操作的状态,也不需要主动的去拷贝数据。

三 I/O 多路复用之select、poll、epoll详解

select,poll,epoll都是IO多路复用的机制。I/O多路复用就是通过一种机制,一个进程可以监视多个描述符,一旦某个描述符就绪(一般是读就绪或者写就绪),能够通知程序进行相应的读写操作。但select,poll,epoll本质上都是同步I/O,因为他们都需要在读写事件就绪后自己负责进行读写,也就是说这个读写过程是阻塞的,而异步I/O则无需自己负责进行读写,异步I/O的实现会负责把数据从内核拷贝到用户空间。(这里啰嗦下)

select

int select (int n, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

select 函数监视的文件描述符分3类,分别是writefds、readfds、和exceptfds。调用后select函数会阻塞,直到有描述副就绪(有数据 可读、可写、或者有except),或者超时(timeout指定等待时间,如果立即返回设为null即可),函数返回。当select函数返回后,可以 通过遍历fdset,来找到就绪的描述符。

select目前几乎在所有的平台上支持,其良好跨平台支持也是它的一个优点。select的一 个缺点在于单个进程能够监视的文件描述符的数量存在最大限制,在Linux上一般为1024,可以通过修改宏定义甚至重新编译内核的方式提升这一限制,但 是这样也会造成效率的降低。

poll

int poll (struct pollfd *fds, unsigned int nfds, int timeout);

不同与select使用三个位图来表示三个fdset的方式,poll使用一个 pollfd的指针实现。

struct pollfd {
    int fd; /* file descriptor */
    short events; /* requested events to watch */
    short revents; /* returned events witnessed */
};

pollfd结构包含了要监视的event和发生的event,不再使用select“参数-值”传递的方式。同时,pollfd并没有最大数量限制(但是数量过大后性能也是会下降)。 和select函数一样,poll返回后,需要轮询pollfd来获取就绪的描述符。

从上面看,select和poll都需要在返回后,通过遍历文件描述符来获取已经就绪的socket。事实上,同时连接的大量客户端在一时刻可能只有很少的处于就绪状态,因此随着监视的描述符数量的增长,其效率也会线性下降。

epoll

epoll是在2.6内核中提出的,是之前的select和poll的增强版本。相对于select和poll来说,epoll更加灵活,没有描述符限制。epoll使用一个文件描述符管理多个描述符,将用户关系的文件描述符的事件存放到内核的一个事件表中,这样在用户空间和内核空间的copy只需一次。

一 epoll操作过程

epoll操作过程需要三个接口,分别如下:

int epoll_create(int size);//创建一个epoll的句柄,size用来告诉内核这个监听的数目一共有多大
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

1. int epoll_create(int size);
创建一个epoll的句柄,size用来告诉内核这个监听的数目一共有多大,这个参数不同于select()中的第一个参数,给出最大监听的fd+1的值,参数size并不是限制了epoll所能监听的描述符最大个数,只是对内核初始分配内部数据结构的一个建议
当创建好epoll句柄后,它就会占用一个fd值,在linux下如果查看/proc/进程id/fd/,是能够看到这个fd的,所以在使用完epoll后,必须调用close()关闭,否则可能导致fd被耗尽。

2. int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
函数是对指定描述符fd执行op操作。
- epfd:是epoll_create()的返回值。
- op:表示op操作,用三个宏来表示:添加EPOLL_CTL_ADD,删除EPOLL_CTL_DEL,修改EPOLL_CTL_MOD。分别添加、删除和修改对fd的监听事件。
- fd:是需要监听的fd(文件描述符)
- epoll_event:是告诉内核需要监听什么事,struct epoll_event结构如下:

struct epoll_event {
  __uint32_t events;  /* Epoll events */
  epoll_data_t data;  /* User data variable */
};

//events可以是以下几个宏的集合:
EPOLLIN :表示对应的文件描述符可以读(包括对端SOCKET正常关闭);
EPOLLOUT:表示对应的文件描述符可以写;
EPOLLPRI:表示对应的文件描述符有紧急的数据可读(这里应该表示有带外数据到来);
EPOLLERR:表示对应的文件描述符发生错误;
EPOLLHUP:表示对应的文件描述符被挂断;
EPOLLET: 将EPOLL设为边缘触发(Edge Triggered)模式,这是相对于水平触发(Level Triggered)来说的。
EPOLLONESHOT:只监听一次事件,当监听完这次事件之后,如果还需要继续监听这个socket的话,需要再次把这个socket加入到EPOLL队列里

3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
等待epfd上的io事件,最多返回maxevents个事件。
参数events用来从内核得到事件的集合,maxevents告之内核这个events有多大,这个maxevents的值不能大于创建epoll_create()时的size,参数timeout是超时时间(毫秒,0会立即返回,-1将不确定,也有说法说是永久阻塞)。该函数返回需要处理的事件数目,如返回0表示已超时。

二 工作模式

 epoll对文件描述符的操作有两种模式:LT(level trigger)ET(edge trigger)。LT模式是默认模式,LT模式与ET模式的区别如下:
  LT模式:当epoll_wait检测到描述符事件发生并将此事件通知应用程序,应用程序可以不立即处理该事件。下次调用epoll_wait时,会再次响应应用程序并通知此事件。
  ET模式:当epoll_wait检测到描述符事件发生并将此事件通知应用程序,应用程序必须立即处理该事件。如果不处理,下次调用epoll_wait时,不会再次响应应用程序并通知此事件。

1. LT模式

LT(level triggered)是缺省的工作方式,并且同时支持block和no-block socket.在这种做法中,内核告诉你一个文件描述符是否就绪了,然后你可以对这个就绪的fd进行IO操作。如果你不作任何操作,内核还是会继续通知你的。

2. ET模式

ET(edge-triggered)是高速工作方式,只支持no-block socket。在这种模式下,当描述符从未就绪变为就绪时,内核通过epoll告诉你。然后它会假设你知道文件描述符已经就绪,并且不会再为那个文件描述符发送更多的就绪通知,直到你做了某些操作导致那个文件描述符不再为就绪状态了(比如,你在发送,接收或者接收请求,或者发送接收的数据少于一定量时导致了一个EWOULDBLOCK 错误)。但是请注意,如果一直不对这个fd作IO操作(从而导致它再次变成未就绪),内核不会发送更多的通知(only once)

ET模式在很大程度上减少了epoll事件被重复触发的次数,因此效率要比LT模式高。epoll工作在ET模式的时候,必须使用非阻塞套接口,以避免由于一个文件句柄的阻塞读/阻塞写操作把处理多个文件描述符的任务饿死。

3. 总结

假如有这样一个例子:
1. 我们已经把一个用来从管道中读取数据的文件句柄(RFD)添加到epoll描述符
2. 这个时候从管道的另一端被写入了2KB的数据
3. 调用epoll_wait(2),并且它会返回RFD,说明它已经准备好读取操作
4. 然后我们读取了1KB的数据
5. 调用epoll_wait(2)......

LT模式:
如果是LT模式,那么在第5步调用epoll_wait(2)之后,仍然能受到通知。

ET模式:
如果我们在第1步将RFD添加到epoll描述符的时候使用了EPOLLET标志,那么在第5步调用epoll_wait(2)之后将有可能会挂起,因为剩余的数据还存在于文件的输入缓冲区内,而且数据发出端还在等待一个针对已经发出数据的反馈信息。只有在监视的文件句柄上发生了某个事件的时候 ET 工作模式才会汇报事件。因此在第5步的时候,调用者可能会放弃等待仍在存在于文件输入缓冲区内的剩余数据。

当使用epoll的ET模型来工作时,当产生了一个EPOLLIN事件后,
读数据的时候需要考虑的是当recv()返回的大小如果等于请求的大小,那么很有可能是缓冲区还有数据未读完,也意味着该次事件还没有处理完,所以还需要再次读取:

while(rs){
  buflen = recv(activeevents[i].data.fd, buf, sizeof(buf), 0);
  if(buflen < 0){
    // 由于是非阻塞的模式,所以当errno为EAGAIN时,表示当前缓冲区已无数据可读
    // 在这里就当作是该次事件已处理处.
    if(errno == EAGAIN){
        break;
    }
    else{
        return;
    }
  }
  else if(buflen == 0){
     // 这里表示对端的socket已正常关闭.
  }

 if(buflen == sizeof(buf){
      rs = 1;   // 需要再次读取
 }
 else{
      rs = 0;
 }
}

Linux中的EAGAIN含义

Linux环境下开发经常会碰到很多错误(设置errno),其中EAGAIN是其中比较常见的一个错误(比如用在非阻塞操作中)。
从字面上来看,是提示再试一次。这个错误经常出现在当应用程序进行一些非阻塞(non-blocking)操作(对文件或socket)的时候。

例如,以 O_NONBLOCK的标志打开文件/socket/FIFO,如果你连续做read操作而没有数据可读。此时程序不会阻塞起来等待数据准备就绪返回,read函数会返回一个错误EAGAIN,提示你的应用程序现在没有数据可读请稍后再试。
又例如,当一个系统调用(比如fork)因为没有足够的资源(比如虚拟内存)而执行失败,返回EAGAIN提示其再调用一次(也许下次就能成功)。

三 代码演示

下面是一段不完整的代码且格式不对,意在表述上面的过程,去掉了一些模板代码。

#define IPADDRESS   "127.0.0.1"
#define PORT        8787
#define MAXSIZE     1024
#define LISTENQ     5
#define FDSIZE      1000
#define EPOLLEVENTS 100

listenfd = socket_bind(IPADDRESS,PORT);

struct epoll_event events[EPOLLEVENTS];

//创建一个描述符
epollfd = epoll_create(FDSIZE);

//添加监听描述符事件
add_event(epollfd,listenfd,EPOLLIN);

//循环等待
for ( ; ; ){
    //该函数返回已经准备好的描述符事件数目
    ret = epoll_wait(epollfd,events,EPOLLEVENTS,-1);
    //处理接收到的连接
    handle_events(epollfd,events,ret,listenfd,buf);
}

//事件处理函数
static void handle_events(int epollfd,struct epoll_event *events,int num,int listenfd,char *buf)
{
     int i;
     int fd;
     //进行遍历;这里只要遍历已经准备好的io事件。num并不是当初epoll_create时的FDSIZE。
     for (i = 0;i < num;i++)
     {
         fd = events[i].data.fd;
        //根据描述符的类型和事件类型进行处理
         if ((fd == listenfd) &&(events[i].events & EPOLLIN))
            handle_accpet(epollfd,listenfd);
         else if (events[i].events & EPOLLIN)
            do_read(epollfd,fd,buf);
         else if (events[i].events & EPOLLOUT)
            do_write(epollfd,fd,buf);
     }
}

//添加事件
static void add_event(int epollfd,int fd,int state){
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    epoll_ctl(epollfd,EPOLL_CTL_ADD,fd,&ev);
}

//处理接收到的连接
static void handle_accpet(int epollfd,int listenfd){
     int clifd;     
     struct sockaddr_in cliaddr;     
     socklen_t  cliaddrlen;     
     clifd = accept(listenfd,(struct sockaddr*)&cliaddr,&cliaddrlen);     
     if (clifd == -1)         
     perror("accpet error:");     
     else {         
         printf("accept a new client: %s:%d\n",inet_ntoa(cliaddr.sin_addr),cliaddr.sin_port);                       //添加一个客户描述符和事件         
         add_event(epollfd,clifd,EPOLLIN);     
     } 
}

//读处理
static void do_read(int epollfd,int fd,char *buf){
    int nread;
    nread = read(fd,buf,MAXSIZE);
    if (nread == -1)     {         
        perror("read error:");         
        close(fd); //记住close fd        
        delete_event(epollfd,fd,EPOLLIN); //删除监听 
    }
    else if (nread == 0)     {         
        fprintf(stderr,"client close.\n");
        close(fd); //记住close fd       
        delete_event(epollfd,fd,EPOLLIN); //删除监听 
    }     
    else {         
        printf("read message is : %s",buf);        
        //修改描述符对应的事件,由读改为写         
        modify_event(epollfd,fd,EPOLLOUT);     
    } 
}

//写处理
static void do_write(int epollfd,int fd,char *buf) {     
    int nwrite;     
    nwrite = write(fd,buf,strlen(buf));     
    if (nwrite == -1){         
        perror("write error:");        
        close(fd);   //记住close fd       
        delete_event(epollfd,fd,EPOLLOUT);  //删除监听    
    }else{
        modify_event(epollfd,fd,EPOLLIN); 
    }    
    memset(buf,0,MAXSIZE); 
}

//删除事件
static void delete_event(int epollfd,int fd,int state) {
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    epoll_ctl(epollfd,EPOLL_CTL_DEL,fd,&ev);
}

//修改事件
static void modify_event(int epollfd,int fd,int state){     
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    epoll_ctl(epollfd,EPOLL_CTL_MOD,fd,&ev);
}

//注:另外一端我就省了

四 epoll总结

在 select/poll中,进程只有在调用一定的方法后,内核才对所有监视的文件描述符进行扫描,而epoll事先通过epoll_ctl()来注册一 个文件描述符,一旦基于某个文件描述符就绪时,内核会采用类似callback的回调机制,迅速激活这个文件描述符,当进程调用epoll_wait() 时便得到通知。(此处去掉了遍历文件描述符,而是通过监听回调的的机制。这正是epoll的魅力所在。)

epoll的优点主要是一下几个方面:
1. 监视的描述符数量不受限制,它所支持的FD上限是最大可以打开文件的数目,这个数字一般远大于2048,举个例子,在1GB内存的机器上大约是10万左 右,具体数目可以cat /proc/sys/fs/file-max察看,一般来说这个数目和系统内存关系很大。select的最大缺点就是进程打开的fd是有数量限制的。这对 于连接数量比较大的服务器来说根本不能满足。虽然也可以选择多进程的解决方案( Apache就是这样实现的),不过虽然linux上面创建进程的代价比较小,但仍旧是不可忽视的,加上进程间数据同步远比不上线程间同步的高效,所以也不是一种完美的方案。

  1. IO的效率不会随着监视fd的数量的增长而下降。epoll不同于select和poll轮询的方式,而是通过每个fd定义的回调函数来实现的。只有就绪的fd才会执行回调函数。

如果没有大量的idle -connection或者dead-connection,epoll的效率并不会比select/poll高很多,但是当遇到大量的idle- connection,就会发现epoll的效率大大高于select/poll。

参考

用户空间与内核空间,进程上下文与中断上下文[总结]
进程切换
维基百科-文件描述符
Linux 中直接 I/O 机制的介绍
IO - 同步,异步,阻塞,非阻塞 (亡羊补牢篇)
Linux中select poll和epoll的区别
IO多路复用之select总结
IO多路复用之poll总结
IO多路复用之epoll总结

查看原文

lxKLj 关注了专栏 · 2月25日

Java核心技术

关注 5093

lxKLj 关注了专栏 · 2月25日

CodeGuide | 程序员编码指南

公众号:bugstack虫洞栈,回复:设计模式,可以下载《重学Java设计模式》PDF,全网下载量17万+ | 这是一本互联网真实案例实践书籍。以落地解决方案为核心,从实际业务中抽离出,交易、营销、秒杀、中间件、源码等22个真实场景,来学习设计模式的运用。

关注 15849

lxKLj 关注了用户 · 2月25日

日拱一兵 @tanrigongyibing

欢迎关注,公众号「日拱一兵」,以读侦探小说思维趣味轻松学习Java技术

关注 23823

lxKLj 关注了专栏 · 2月25日

Java进阶课

关注公众号:【程序员内点事】,架构技术、面试资料持续分享

关注 1833

lxKLj 发布了文章 · 2月25日

ThreadPoolTaskScheduler 周期任务原理

ThreadPoolTaskScheduler 核心就是schedule 方法

public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
    ScheduledExecutorService executor = getScheduledExecutor();
    try {
        ErrorHandler errorHandler = this.errorHandler;
        if (errorHandler == null) {
            errorHandler = TaskUtils.getDefaultErrorHandler(true);
        }
      // 最终调用ReschedulingRunnable.schedule 方法
        return new ReschedulingRunnable(task, trigger, executor, errorHandler).schedule();
    }
    catch (RejectedExecutionException ex) {
        throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
    }
}

后续进入ReschedulingRunnable.schedule 方法,该类中executor 属性为ScheduledThreadPoolExecutor 类,属性为ScheduledThreadPoolExecutor 类继承了ThreadPoolExecutor 线程池,但是自定义了DelayedWorkQueue 延迟队列,而不是使用ThreadPoolExecutor 类自带的队列,周期任务延迟执行的根本原因就是DelayedWorkQueue 这个延迟队列。

public ScheduledFuture<?> schedule() {
    synchronized (this.triggerContextMonitor) {
        this.scheduledExecutionTime = this.trigger.nextExecutionTime(this.triggerContext);
        if (this.scheduledExecutionTime == null) {
            return null;
        }
        long initialDelay = this.scheduledExecutionTime.getTime() - System.currentTimeMillis();
        this.currentFuture = this.executor.schedule(this, initialDelay, TimeUnit.MILLISECONDS);
        return this;
    }
}

DelayedWorkQueue 类通过一个最小堆来存储ThreadPoolTaskScheduler 中的任务,各任务会进行比较,最快要执行的任务放在最小堆顶部。放入最小堆,通过siftUp,取出最小堆,通过siftDown。

/**
 * 上浮
 */
private void siftUp(int k, RunnableScheduledFuture<?> key) {
    // 一直遍历到根节点下方
    while (k > 0) {
        // 二叉堆,最高节点坐标为0
        // 父节点,(k - 1)/2
        int parent = (k - 1) >>> 1;
        RunnableScheduledFuture<?> e = queue[parent];
        // 目标比父节点大
        // 不需要再上浮,直接跳出
        if (key.compareTo(e) >= 0)
            break;
        // 目标节点比父节点小
        // 继续上浮,当前坐标填入父节点
        queue[k] = e;
        setIndex(e, k);
        // 当前坐标设为父节点坐标
        k = parent;
    }
    // 目标上浮到最小节点坐标,填入该坐标
    queue[k] = key;
    setIndex(key, k);
}

/**
 * 下沉
 */
private void siftDown(int k, RunnableScheduledFuture<?> key) {
    // half = size/2;
    // 二叉堆,最高节点坐标为0
    // 任何节点,其左子节点坐标(k*2)+1,右子节点坐标(k*2)+2
    int half = size >>> 1;
    // 需要拿到子节点,所以只需要到size/2 即可,不需要到最底层
    while (k < half) {
        // 左节点坐标
        int child = (k << 1) + 1;
        RunnableScheduledFuture<?> c = queue[child];
        // 右节点坐标
        int right = child + 1;
        // 左节点比右节点大
        if (right < size && c.compareTo(queue[right]) > 0)
            // 最小 = 右子节点
            c = queue[child = right];
        // 目标最小比子节点小
        // 不再需要下沉,直接退出,目标填入当前坐标
        if (key.compareTo(c) <= 0)
            break;
        // 目标比子节点大
        // 继续下沉,小子节点放到当前节点坐标
        queue[k] = c;
        setIndex(c, k);
        // 当前坐标设为子节点坐标
        // 坐标不断下沉
        k = child;
    }
    // 当前坐标为目标对象最小情况下的坐标
    // 讲目标对象放入该坐标
    queue[k] = key;
    setIndex(key, k);
}

当从DelayedWorkQueue 队列中取出任务时,会取出最小堆顶部的任务,也就是最快要执行的任务,然后线程等待指定时间。等待时间结束后,通过自旋完成任务。

public RunnableScheduledFuture<?> take() throws InterruptedException {
    ...
    for (;;) {
        RunnableScheduledFuture<?> first = queue[0];
        ...
        if (delay <= 0)
            return finishPoll(first);
        ...
        available.awaitNanos(delay);
        ...
    }
   ...
}
查看原文

赞 0 收藏 0 评论 0

lxKLj 发布了文章 · 2020-01-06

单链表的头插法和尾插法

背景

在看HashMap 的源码时,发现扩容操作中对于链表的复制操作,不是很能理解,代码如下所示:

Node<K,V> loHead = null, loTail = null;
Node<K,V> hiHead = null, hiTail = null;
Node<K,V> next;
do {
    next = e.next;
    if ((e.hash & oldCap) == 0) {
        if (loTail == null)
            loHead = e;
        else
            loTail.next = e;
        loTail = e;
    }
    else {
        if (hiTail == null)
            hiHead = e;
        else
            hiTail.next = e;
        hiTail = e;
    }
} while ((e = next) != null);

之后再网上查了一下,发现这是一个典型的单链表的尾插法操作,后续对单链表的头插法也进行了相关了解,整理了这片文章,与大家一起分享一下。

头插法

顾名思义,头插法就是在单链表的节点插入操作中,新的节点总是在前面,结果有点类似栈的先进后出。下面是一段头插法的代码实现:

public class Node {

    /**
     * 节点存的内容
     */
    private Integer value;

    /**
     * 链表的下一个节点
     */
    private Node next;

    /**
     * 用于记录所有生成的值
     * 比较头插法 和 尾插法对于数据的顺序影响
     */
    public static Integer[] valueArray;

    /**
     * 生成指定数量的节点链表
     *
     * @param count     链表节点数
     * @return          链表节点
     */
    public static Node generateLinkedNode(int count) {
        Node head = null;
        // 初始化记录数组
        valueArray = new Integer[count];
        for (int i = 0; i < count; i++) {
            // 新节点
            Node newNode = new Node();
            // 随机一个值
            Integer value = new Random().nextInt(100) + 1;
            // 新节点赋值
            newNode.value = value;
            // 记录节点存的值
            valueArray[i] = value;
            if (head == null) {
                // 头结点不存在时,新节点赋值成头结点
                head = newNode;
            } else {
                // 存在头结点,将新节点的下一节点设置成头结点
                // 也就是说在头结点前插入新节点
                newNode.next = head;
                // 将新节点变成头结点
                // 与上一步相呼应,移动链表
                head = newNode;
            }
        }
        return head;
    }

    public Integer getValue() {
        return value;
    }

    public Node getNext() {
        return next;
    }
}
public static void main(String[] args) {
    Node linkedNode = Node.generateLinkedNode(5);
    System.out.println(Arrays.toString(Node.valueArray));
    do {
        if (linkedNode.getNext() == null) {
            System.out.print(linkedNode.getValue());
        } else {
            System.out.print(linkedNode.getValue() + ", ");
        }
    } while ((linkedNode = linkedNode.getNext()) != null);
}

结果如下所示,我们可以看到,数据的生成的顺序是63,78,85,62,60,但是我们采用头插法插入后,遍历节点取出数据,发现数据恰恰是相反的顺序,这是什么原因呢?
image.png


现在就让我们通过画图来更明确地展示整个头插法的流程,来回答上面这个问题。

if (head == null) {
    // 头结点不存在时,新节点赋值成头结点
    head = newNode;
} 

这段代码,我们可以看出,当头结点不存在时,我们会把插入的这个节点作为头结点。如下图所示:
image.png

// 存在头结点,将新节点的下一节点设置成头结点
// 也就是说在头结点前插入新节点
newNode.next = head;
// 将新节点变成头结点
// 与上一步相呼应,移动链表
head = newNode;

头结点存在时,我们就需要将新节点的下一节点设置成头结点。如下图所示:
image.png
后面新插入的节点,不断地重复这一步,直到没有新节点为止,最终的单链表结构就如下所示:
image.png
这就是单链表的头插法,新节点总是在链表的头部。


尾插法

尾插法,也比较好理解,就是每一个新节点都是插入到链表的尾部,有点类似于队列的先进先出。下面来看看尾插法的实现:

/**
 * 生成指定数量的节点链表-尾插法
 *
 * @param count     链表节点数
 * @return          链表节点
 */
public static Node generateTailLinkedNode(int count) {
    // 头结点不能移动,所以需要一个临时节点来进行操作
    Node head = null, temp = null;
    // 初始化记录数组
    valueArray = new Integer[count];
    for (int i = 0; i < count; i++) {
        // 新节点
        Node newNode = new Node();
        // 随机一个值
        Integer value = new Random().nextInt(100) + 1;
        // 新节点赋值
        newNode.value = value;
        // 记录节点存的值
        valueArray[i] = value;
        if (temp == null) {
            // 头结点不存在时,新节点赋值成头结点
            // 这一步结合下面的temp = newNode,两者此时都指向同一节点,
            // 所以后续对temp 节点的操作其实就是对与head 的这一节点操作。
            head = newNode;
        } else {
            // 存在头结点,将新节点设置为下一节点
            // 也就是说在头结点尾部插入新节点
            temp.next = newNode;
        }
        // 临时节点重新赋值,移动到下一节点
        temp = newNode;
    }
    return head;
}

我们运行同样的main 方法,发现新节点的插入顺序是和输出顺序一样的,也就是说每一个新节点都是插入上一节点的尾部。
image.png
下面,我们来根据代码,一步步地画图来看看这是如何实现的,
第一步和头插法是一样的,我们就不细看了,主要是看下面的操作。

if (temp == null) {
    // 头结点不存在时,新节点赋值成头结点
    // 这一步结合下面的temp = newNode,两者此时都指向同一节点,
    // 所以后续对temp 节点的操作其实就是对与head 的这一节点操作。
    head = newNode;
} else {
    // 存在头结点,将新节点设置为下一节点
    // 也就是说在头结点尾部插入新节点
    temp.next = newNode;
}
// 临时节点重新赋值,移动到下一节点
temp = newNode;

在头结点存在的情况下,新插入一个节点,我们需要将新节点设置为下一节点,并且最终要移动临时节点指向下一节点。
image.png
对于后续的节点插入,我们只需要不断地重复这一操作即可,temp 临时节点在没插入一个新节点后,都需要重新指向这个新节点,为下一节点的插入做准备。最终我们的链表就是如下所示的情况:
image.png
这就是单链表的尾插法,其头结点就是我们的第一个节点,后续的节点都是插入上一节点的尾部,最终的输出顺序就是我们输入的节点顺序。


以上就是单链表插入的两种方法,希望对大家能有所帮助,能够更好地理解头插法和尾插法。

查看原文

赞 0 收藏 0 评论 0

lxKLj 关注了用户 · 2019-12-30

阿里云云栖号 @yunqishequ_5aa899aad5395

阿里云官网内容平台!汇聚阿里云优质内容(入门、文档、案例、最佳实践、直播等)!如需转载或内容类合作,邮件yqgroup@service.aliyun.com 秒级回复!

关注 10938

lxKLj 发布了文章 · 2019-12-30

HashMap(1.8) 源码理解

目前正在研究HashMap 的源码,写点东西做记录,欢迎大家相互学习,也欢迎大佬进行指点。

  • HashMap 静态变量
  • HashMap 节点
  • HashMap put 操作

HashMap 静态变量

/**
 * Map 默认的大小 16。
 */
static final int DEFAULT_INITIAL_CAPACITY = 1 << 4;

/**
 * Map 的最大存储量。
 */
static final int MAXIMUM_CAPACITY = 1 << 30;

/**
 * Map 因子,CAPACITY * LOAD_FACTOR 用来决定何时需要扩容。
 * 为啥是0.75,有待考究,如果有大佬能解惑,希望不吝指教。
 */
static final float DEFAULT_LOAD_FACTOR = 0.75f;

/**
 * 确定转换为红黑树的阈值
 */
static final int TREEIFY_THRESHOLD = 8;

/**
 * 确定红黑树回退的阈值
 */
static final int UNTREEIFY_THRESHOLD = 6;

/**
 * 可以转换成红黑树的数组最小值
 */
static final int MIN_TREEIFY_CAPACITY = 64;

HashMap 节点

Node 节点

HashMap 底层对于数据的存储实际上是以Node 节点(红黑树为TreeNode)的方式进行存储。我们进行put 操作的key-value 则是存入Node 节点中,可以说HashMap 底层的最小单位就是节点。

/**
 * Map 内部自己的hash 值计算方法
 */
static final int hash(Object key) {
    int h;
    return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
}

/**
 * Map 实际的存储对象
 */
static class Node<K,V> implements Map.Entry<K,V> {
    /**
     * hash 方法计算出的值
     */
    final int hash;
    /**
     * Map 存入的key
     */
    final K key;
    /**
     * Map 存入的value
     */
    V value;
    /**
     * 链表结构,hash 冲突且key 不同时存入
     */
    Node<K,V> next;

    Node(int hash, K key, V value, Node<K,V> next) {
        this.hash = hash;
        this.key = key;
        this.value = value;
        this.next = next;
    }

    public final K getKey()        { return key; }
    public final V getValue()      { return value; }
    public final String toString() { return key + "=" + value; }

    public final int hashCode() {
        return Objects.hashCode(key) ^ Objects.hashCode(value);
    }

    public final V setValue(V newValue) {
        V oldValue = value;
        value = newValue;
        return oldValue;
    }

    public final boolean equals(Object o) {
        if (o == this)
            return true;
        if (o instanceof Map.Entry) {
            Map.Entry<?,?> e = (Map.Entry<?,?>)o;
            if (Objects.equals(key, e.getKey()) &&
                Objects.equals(value, e.getValue()))
                return true;
        }
        return false;
    }
}

TreeNode 节点

当链表存储的过大,且满足转换成红黑树的条件时,链表会被转换成红黑树。此时节点会重新生成为TreeNode ,当然看源码我们可以知道,TreeNode 实际上还是继承的Node 节点。

/**
 * Map 红黑树节点
 */
static final class TreeNode<K,V> extends LinkedHashMap.Entry<K,V> {
    /**
    * 父节点
    */
    TreeNode<K,V> parent;  
    /**
    * 左节点
    */
    TreeNode<K,V> left;
    /**
    * 右节点
    */
    TreeNode<K,V> right;
    /**
    * 需要进一步研究红黑树算法
    */
    TreeNode<K,V> prev;   
    boolean red;
    TreeNode(int hash, K key, V val, Node<K,V> next) {
        super(hash, key, val, next);
    }

    /**
     * 返回红黑树的根节点
     */
    final TreeNode<K,V> root() {
        for (TreeNode<K,V> r = this, p;;) {
            if ((p = r.parent) == null)
                return r;
            r = p;
        }
    }
    
    // 省略
    ...
}

HashMap put 操作

put 操作主要分为两种情况,hash 冲突和不冲突。

  • hash 冲突

hash 冲突,即(n - 1) & hash定位到的数组坐标,已经有值存在。此时需要对key 值再进行判断,如果key 相同,则覆盖旧值;key 不同,则需要将节点放入链表尾部,或者若是红黑树节点,则放入红黑树中。

  • hash 不冲突

hash 不冲突,则只需直接将节点放入数组中即可。

/**
 * Map 存放数据
 *
 * @param hash    map 自有算法((h = key.hashCode()) ^ (h >>> 16),
 * key.hashCode()为key 类的hashCode 实现,如String 为s[0]*31^(n-1) + s[1]*31^(n-2) + ... + s[n-1],n 为字符串长度,s[i] 为i 位置字符)
 * @param key    要存入map 的key
 * @param value    要存入map 的value
 * @param onlyIfAbsent 是否覆盖
 * @param evict
 * @return        如果覆盖,则为覆盖前的值
 */
final V putVal(int hash, K key, V value, boolean onlyIfAbsent, boolean evict) {
    Node<K,V>[] tab; Node<K,V> p; int n, i;
    //判断数组是否为空,如果为空则初始化数组
    if ((tab = table) == null || (n = tab.length) == 0)
        n = (tab = resize()).length;
    //判断当前位置是否有值
    //没值,则放入新Node(hash,key,value,Node next) 节点
    if ((p = tab[i = (n - 1) & hash]) == null)
        tab[i] = newNode(hash, key, value, null);
    else {
        //hash冲突,当前位置有值
        Node<K,V> e; K k;
        //判断这俩个key 是否一样
        if (p.hash == hash &&
            ((k = p.key) == key || (key != null && key.equals(k))))
            //一样,则把当前节点p 赋值给临时节点e
            e = p;
        //key 不一样,新节点需要放入链表,如果链表已转成红黑树,则放入红黑树中
        //判断节点是否为红黑树节点
        else if (p instanceof TreeNode)
            //红黑树节点,将新节点放入红黑树中
            e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value);
        else {
            //链表
            for (int binCount = 0; ; ++binCount) {
                //定位到链表尾端
                if ((e = p.next) == null) {
                    //将新节点插入链表尾端
                    p.next = newNode(hash, key, value, null);
                    //判断是否需要将链表转换成红黑树结构
                    if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st
                        treeifyBin(tab, hash);
                    break;
                }
                //如果在链表中找到key 相同的,也就是说链表中已存在,则跳出
                if (e.hash == hash &&
                    ((k = e.key) == key || (key != null && key.equals(k))))
                    break;
                //遍历链表,呼应上文e = p.next
                p = e;
            }
        }
        //判断是否已存在
        if (e != null) { // existing mapping for key
            V oldValue = e.value;
            //判断是否需要覆盖
            if (!onlyIfAbsent || oldValue == null)
                //覆盖值
                e.value = value;
            afterNodeAccess(e);
            //返回旧值
            return oldValue;
        }
    }
    //map 修改次数
    ++modCount;
    //判断Map 是否需要扩容 
    if (++size > threshold)
        resize();
    afterNodeInsertion(evict);
    return null;
}

未完待续。。。

查看原文

赞 1 收藏 0 评论 0

认证与成就

  • 获得 1 次点赞
  • 获得 0 枚徽章 获得 0 枚金徽章, 获得 0 枚银徽章, 获得 0 枚铜徽章

擅长技能
编辑

开源项目 & 著作
编辑

(゚∀゚ )
暂时没有

注册于 2019-12-19
个人主页被 506 人浏览