What hurts more, the pain of hard work or the pain of regret?

0%

前面说了这么些数据结构,其实大家对于 redis 最初的印象应该就是个 key-value 的缓存,类似于 memcache,redis 其实也是个 key-value,key 还是一样的字符串,或者说就是用 redis 自己的动态字符串实现,但是 value 其实就是前面说的那些数据结构,差不多快说完了,还有个 quicklist 后面还有一篇,这里先介绍下 redis 对于这些不同类型的 value 是怎么实现的,首先看下 redisObject 的源码头文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
/* The actual Redis Object */
#define OBJ_STRING 0 /* String object. */
#define OBJ_LIST 1 /* List object. */
#define OBJ_SET 2 /* Set object. */
#define OBJ_ZSET 3 /* Sorted set object. */
#define OBJ_HASH 4 /* Hash object. */
/*
* Objects encoding. Some kind of objects like Strings and Hashes can be
* internally represented in multiple ways. The 'encoding' field of the object
* is set to one of this fields for this object. */
#define OBJ_ENCODING_RAW 0 /* Raw representation */
#define OBJ_ENCODING_INT 1 /* Encoded as integer */
#define OBJ_ENCODING_HT 2 /* Encoded as hash table */
#define OBJ_ENCODING_ZIPMAP 3 /* Encoded as zipmap */
#define OBJ_ENCODING_LINKEDLIST 4 /* No longer used: old list encoding. */
#define OBJ_ENCODING_ZIPLIST 5 /* Encoded as ziplist */
#define OBJ_ENCODING_INTSET 6 /* Encoded as intset */
#define OBJ_ENCODING_SKIPLIST 7 /* Encoded as skiplist */
#define OBJ_ENCODING_EMBSTR 8 /* Embedded sds string encoding */
#define OBJ_ENCODING_QUICKLIST 9 /* Encoded as linked list of ziplists */
#define OBJ_ENCODING_STREAM 10 /* Encoded as a radix tree of listpacks */

#define LRU_BITS 24
#define LRU_CLOCK_MAX ((1<<LRU_BITS)-1) /* Max value of obj->lru */
#define LRU_CLOCK_RESOLUTION 1000 /* LRU clock resolution in ms */

#define OBJ_SHARED_REFCOUNT INT_MAX
typedef struct redisObject {
unsigned type:4;
unsigned encoding:4;
unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
* LFU data (least significant 8 bits frequency
* and most significant 16 bits access time). */
int refcount;
void *ptr;
} robj;

主体结构就是这个 redisObject,

  • type: 字段表示对象的类型,它对应的就是 redis 的对外暴露的,或者说用户可以使用的五种类型,OBJ_STRING, OBJ_LIST, OBJ_SET, OBJ_ZSET, OBJ_HASH
  • encoding: 字段表示这个对象在 redis 内部的编码方式,由OBJ_ENCODING_开头的 11 种
  • lru: 做LRU替换算法用,占24个bit
  • refcount: 引用计数。它允许robj对象在某些情况下被共享。
  • ptr: 指向底层实现数据结构的指针
    当 type 是 OBJ_STRING 时,表示类型是个 string,它的编码方式 encoding 可能有 OBJ_ENCODING_RAW,OBJ_ENCODING_INT,OBJ_ENCODING_EMBSTR 三种
    当 type 是 OBJ_LIST 时,表示类型是 list,它的编码方式 encoding 是 OBJ_ENCODING_QUICKLIST,对于早一些的版本,2.2这种可能还会使用 OBJ_ENCODING_ZIPLIST,OBJ_ENCODING_LINKEDLIST
    当 type 是 OBJ_SET 时,是个集合,但是得看具体元素的类型,有可能使用整数集合,OBJ_ENCODING_INTSET, 如果元素不全是整型或者数量超过一定限制,那么编码就是 OBJ_ENCODING_HT hash table 了
    当 type 是 OBJ_ZSET 时,是个有序集合,它底层有可能使用的是 OBJ_ENCODING_ZIPLIST 或者 OBJ_ENCODING_SKIPLIST
    当 type 是 OBJ_HASH 时,一开始也是 OBJ_ENCODING_ZIPLIST,然后当数据量大于 hash_max_ziplist_entries 时会转成 OBJ_ENCODING_HT

在 redis 中还有一类表型数据结构叫压缩表,ziplist,它的目的是替代链表,链表是个很容易理解的数据结构,双向链表有前后指针,有带头结点的有的不带,但是链表有个比较大的问题是相对于普通的数组,它的内存不连续,碎片化的存储,内存利用效率不高,而且指针寻址相对于直接使用偏移量的话,也有一定的效率劣势,当然这不是主要的原因,ziplist 设计的主要目的是让链表的内存使用更高效

The ziplist is a specially encoded dually linked list that is designed to be very memory efficient.
这是摘自 redis 源码中ziplist.c 文件的注释,也说明了原因,它的大概结构是这样子

1
<zlbytes> <zltail> <zllen> <entry> <entry> ... <entry> <zlend>

其中
<zlbytes>表示 ziplist 占用的字节总数,类型是uint32_t,32 位的无符号整型,当然表示的字节数也包含自己本身占用的 4 个
<zltail> 类型也是是uint32_t,表示ziplist表中最后一项(entry)在ziplist中的偏移字节数。<zltail>的存在,使得我们可以很方便地找到最后一项(不用遍历整个ziplist),从而可以在ziplist尾端快速地执行push或pop操作。
<uint16_t zllen> 表示ziplist 中的数据项个数,因为是 16 位,所以当数量超过所能表示的最大的数量,它的 16 位全会置为 1,但是真实的数量需要遍历整个 ziplist 才能知道
<entry>是具体的数据项,后面解释
<zlend> ziplist 的最后一个字节,固定是255。
再看一下<entry>中的具体结构,

1
<prevlen> <encoding> <entry-data>

首先这个<prevlen>有两种情况,一种是前面的元素的长度,如果是小于等于 253的时候就用一个uint8_t 来表示前一元素的长度,如果大于的话他将占用五个字节,第一个字节是 254,即表示这个字节已经表示不下了,需要后面的四个字节帮忙表示
<encoding>这个就比较复杂,把源码的注释放下面先看下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
* |00pppppp| - 1 byte
* String value with length less than or equal to 63 bytes (6 bits).
* "pppppp" represents the unsigned 6 bit length.
* |01pppppp|qqqqqqqq| - 2 bytes
* String value with length less than or equal to 16383 bytes (14 bits).
* IMPORTANT: The 14 bit number is stored in big endian.
* |10000000|qqqqqqqq|rrrrrrrr|ssssssss|tttttttt| - 5 bytes
* String value with length greater than or equal to 16384 bytes.
* Only the 4 bytes following the first byte represents the length
* up to 32^2-1. The 6 lower bits of the first byte are not used and
* are set to zero.
* IMPORTANT: The 32 bit number is stored in big endian.
* |11000000| - 3 bytes
* Integer encoded as int16_t (2 bytes).
* |11010000| - 5 bytes
* Integer encoded as int32_t (4 bytes).
* |11100000| - 9 bytes
* Integer encoded as int64_t (8 bytes).
* |11110000| - 4 bytes
* Integer encoded as 24 bit signed (3 bytes).
* |11111110| - 2 bytes
* Integer encoded as 8 bit signed (1 byte).
* |1111xxxx| - (with xxxx between 0000 and 1101) immediate 4 bit integer.
* Unsigned integer from 0 to 12. The encoded value is actually from
* 1 to 13 because 0000 and 1111 can not be used, so 1 should be
* subtracted from the encoded 4 bit value to obtain the right value.
* |11111111| - End of ziplist special entry.

首先如果 encoding 的前两位是 00 的话代表这个元素是个 6 位的字符串,即直接将数据保存在 encoding 中,不消耗额外的<entry-data>,如果前两位是 01 的话表示是个 14 位的字符串,如果是 10 的话表示encoding 块之后的四个字节是存放字符串类型的数据,encoding 的剩余 6 位置 0。
如果 encoding 的前两位是 11 的话表示这是个整型,具体的如果后两位是00的话,表示后面是个2字节的 int16_t 类型,如果是01的话,后面是个4字节的int32_t,如果是10的话后面是8字节的int64_t,如果是 11 的话后面是 3 字节的有符号整型,这些都要最后 4 位都是 0 的情况噢
剩下当是11111110时,则表示是一个1 字节的有符号数,如果是 1111xxxx,其中xxxx在0000 到 1101 表示实际的 1 到 13,为啥呢,因为 0000 前面已经用过了,而 1110 跟 1111 也都有用了。
看个具体的例子(上下有点对不齐,将就看)

1
2
[0f 00 00 00] [0c 00 00 00] [02 00] [00 f3] [02 f6] [ff]
|**zlbytes***| |***zltail***| |*zllen*| |entry1 entry2| |zlend|

第一部分代表整个 ziplist 有 15 个字节,zlbytes 自己占了 4 个 zltail 表示最后一个元素的偏移量,第 13 个字节起,zllen 表示有 2 个元素,第一个元素是00f3,00表示前一个元素长度是 0,本来前面就没元素(不过不知道这个能不能优化这一字节),然后是 f3,换成二进制就是11110011,对照上面的注释,是落在|1111xxxx|这个类型里,注意这个其实是用 0001 到 1101 也就是 1到 13 来表示 0到 12,所以 f3 应该就是 2,第一个元素是 2,第二个元素呢,02 代表前一个元素也就是刚才说的这个,占用 2 字节,f6 展开也是刚才的类型,实际是 5,ff 表示 ziplist 的结尾,所以这个 ziplist 里面是两个元素,2 跟 5

redis中对于 set 其实有两种处理,对于元素均为整型,并且元素数目较少时,使用 intset 作为底层数据结构,否则使用 dict 作为底层数据结构,先看一下代码先

1
2
3
4
5
6
7
8
9
10
11
12
13
14
typedef struct intset {
// 编码方式
uint32_t encoding;
// 集合包含的元素数量
uint32_t length;
// 保存元素的数组
int8_t contents[];
} intset;

/* Note that these encodings are ordered, so:
* INTSET_ENC_INT16 < INTSET_ENC_INT32 < INTSET_ENC_INT64. */
#define INTSET_ENC_INT16 (sizeof(int16_t))
#define INTSET_ENC_INT32 (sizeof(int32_t))
#define INTSET_ENC_INT64 (sizeof(int64_t))

一眼看,为啥整型还需要编码,然后 int8_t 怎么能存下大整形呢,带着这些疑问,我们一步步分析下去,这里的编码其实指的是这个整型集合里存的究竟是多大的整型,16 位,还是 32 位,还是 64 位,结构体下面的宏定义就是表示了 encoding 的可能取值,INTSET_ENC_INT16 表示每个元素用2个字节存储,INTSET_ENC_INT32 表示每个元素用4个字节存储,INTSET_ENC_INT64 表示每个元素用8个字节存储。因此,intset中存储的整数最多只能占用64bit。length 就是正常的表示集合中元素的数量。最奇怪的应该就是这个 contents 了,是个 int8_t 的数组,那放毛线数据啊,最小的都有 16 位,这里我在看代码和《redis 设计与实现》的时候也有点懵逼,后来查了下发现这是个比较取巧的用法,这里我用自己的理解表述一下,先看看 8,16,32,64 的关系,一眼看就知道都是 2 的 N 次,并且呈两倍关系,而且 8 位刚好一个字节,所以呢其实这里的contents 不是个常规意义上的 int8_t 类型的数组,而是个柔性数组。看下 wiki 的定义

Flexible array members1 were introduced in the C99 standard of the C programming language (in particular, in section §6.7.2.1, item 16, page 103).2 It is a member of a struct, which is an array without a given dimension. It must be the last member of such a struct and it must be accompanied by at least one other member, as in the following example:

1
2
3
4
struct vectord {
size_t len;
double arr[]; // the flexible array member must be last
};

在初始化这个 intset 的时候,这个contents数组是不占用空间的,后面的反正用到了申请,那么这里就有一个问题,给出了三种可能的 encoding 值,他们能随便换吗,显然不行,首先在 intset 中数据的存放是有序的,这个有部分原因是方便二分查找,然后存放数据其实随着数据的大小不同会有一个升级的过程,看下图

新创建的intset只有一个header,总共8个字节。其中encoding = 2, length = 0, 类型都是uint32_t,各占 4 字节,添加15, 5两个元素之后,因为它们是比较小的整数,都能使用2个字节表示,所以encoding不变,值还是2,也就是默认的 INTSET_ENC_INT16,当添加32768的时候,它不再能用2个字节来表示了(2个字节能表达的数据范围是-215~215-1,而32768等于215,超出范围了),因此encoding必须升级到INTSET_ENC_INT32(值为4),即用4个字节表示一个元素。在添加每个元素的过程中,intset始终保持从小到大有序。与ziplist类似,intset也是按小端(little endian)模式存储的(参见维基百科词条Endianness)。比如,在上图中intset添加完所有数据之后,表示encoding字段的4个字节应该解释成0x00000004,而第4个数据应该解释成0x00008000 = 32768

跳表 skiplist

跳表是个在我们日常的代码中不太常用到的数据结构,相对来讲就没有像数组,链表,字典,散列,树等结构那么熟悉,所以就从头开始分析下,首先是链表,跳表跟链表都有个表字(太硬扯了我🤦‍♀️),注意这是个有序链表

如上图,在这个链表里如果我要找到 23,是不是我需要从3,5,9开始一直往后找直到找到 23,也就是说时间复杂度是 O(N),N 的一次幂复杂度,那么我们来看看第二个

这个结构跟原先有点不一样,它给链表中偶数位的节点又加了一个指针把它们链接起来,这样子当我们要寻找 23 的时候就可以从原来的一个个往下找变成跳着找,先找到 5,然后是 10,接着是 19,然后是 28,这时候发现 28 比 23 大了,那我在退回到 19,然后从下一层原来的链表往前找,

这里毛估估是不是前面的节点我就少找了一半,有那么点二分法的意思。
前面的其实是跳表的引子,真正的跳表其实不是这样,因为上面的其实有个比较大的问题,就是插入一个元素后需要调整每个元素的指针,在 redis 中的跳表其实是做了个随机层数的优化,因为沿着前面的例子,其实当数据量很大的时候,是不是层数越多,其查询效率越高,但是随着层数变多,要保持这种严格的层数规则其实也会增大处理复杂度,所以 redis 插入每个元素的时候都是使用随机的方式,看一眼代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
/* ZSETs use a specialized version of Skiplists */
typedef struct zskiplistNode {
sds ele;
double score;
struct zskiplistNode *backward;
struct zskiplistLevel {
struct zskiplistNode *forward;
unsigned long span;
} level[];
} zskiplistNode;

typedef struct zskiplist {
struct zskiplistNode *header, *tail;
unsigned long length;
int level;
} zskiplist;

typedef struct zset {
dict *dict;
zskiplist *zsl;
} zset;

忘了说了,redis 是把 skiplist 跳表用在 zset 里,zset 是个有序的集合,可以看到 zskiplist 就是个跳表的结构,里面用 header 保存跳表的表头,tail 保存表尾,还有长度和最大层级,具体的跳表节点元素使用 zskiplistNode 表示,里面包含了 sds 类型的元素值,double 类型的分值,用来排序,一个 backward 后向指针和一个 zskiplistLevel 数组,每个 level 包含了一个前向指针,和一个 span,span 表示的是跳表前向指针的跨度,这里再补充一点,前面说了为了灵活这个跳表的新增修改,redis 使用了随机层高的方式插入新节点,但是如果所有节点都随机到很高的层级或者所有都很低的话,跳表的效率优势就会减小,所以 redis 使用了个小技巧,贴下代码

1
2
3
4
5
6
7
#define ZSKIPLIST_P 0.25      /* Skiplist P = 1/4 */
int zslRandomLevel(void) {
int level = 1;
while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
level += 1;
return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

当随机值跟0xFFFF进行与操作小于ZSKIPLIST_P * 0xFFFF时才会增大 level 的值,因此保持了一个相对递减的概率
可以简单分析下,当 random() 的值小于 0xFFFF 的 1/4,才会 level + 1,就意味着当有 1 - 1/4也就是3/4的概率是直接跳出,所以一层的概率是3/4,也就是 1-P,二层的概率是 P*(1-P),三层的概率是 P² * (1-P) 依次递推。

redis是现在服务端很常用的缓存中间件,其实原来还有memcache之类的竞品,但是现在貌似 redis 快一统江湖,这里当然不是在吹,只是个人角度的一个感觉,不权威只是主观感觉。
redis 主要有五种数据结构,StringsListsSetsHashesSorted Sets,这五种数据结构先简单介绍下,Strings类型的其实就是我们最常用的 key-value,实际开发中也会用的最多;Lists是列表,这个有些会用来做队列,因为 redis 目前常用的版本支持丰富的列表操作;还有是Sets集合,这个主要的特点就是集合中元素不重复,可以用在有这类需求的场景里;Hashes是叫散列,类似于 Python 中的字典结构;还有就是Sorted Sets这个是个有序集合;一眼看这些其实没啥特别的,除了最后这个有序集合,不过去了解背后的实现方式还是比较有意思的。

SDS 简单动态字符串

先从Strings开始说,了解过 C 语言的应该知道,C 语言中的字符串其实是个 char[] 字符数组,redis 也不例外,只是最开始的版本就对这个做了一丢丢的优化,而正是这一丢丢的优化,让这个 redis 的使用效率提升了数倍

1
2
3
4
5
6
7
8
struct sdshdr {
// 字符串长度
int len;
// 字符串空余字符数
int free;
// 字符串内容
char buf[];
};

这里引用了 redis 在 github 上最早的 2.2 版本的代码,代码路径是https://github.com/antirez/redis/blob/2.2/src/sds.h,可以看到这个结构体里只有仨元素,两个 int 型和一个 char 型数组,两个 int 型其实就是我说的优化,因为 C 语言本身的字符串数组,有两个问题,一个是要知道它实际已被占用的长度,需要去遍历这个数组,第二个就是比较容易踩坑的是遍历的时候要注意它有个以\0作为结尾的特点;通过上面的两个 int 型参数,一个是知道字符串目前的长度,一个是知道字符串还剩余多少位空间,这样子坐着两个操作从 O(N)简化到了O(1)了,还有第二个 free 还有个比较重要的作用就是能防止 C 字符串的溢出问题,在存储之前可以先判断 free 长度,如果长度不够就先扩容了,先介绍到这,这个系列可以写蛮多的,慢慢介绍吧

链表

链表是比较常见的数据结构了,但是因为 redis 是用 C 写的,所以在不依赖第三方库的情况下只能自己写一个了,redis 的链表是个有头的链表,而且是无环的,具体的结构我也找了 github 上最早版本的代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
typedef struct listNode {
// 前置节点
struct listNode *prev;
// 后置节点
struct listNode *next;
// 值
void *value;
} listNode;

typedef struct list {
// 链表表头
listNode *head;
// 当前节点,也可以说是最后节点
listNode *tail;
// 节点复制函数
void *(*dup)(void *ptr);
// 节点值释放函数
void (*free)(void *ptr);
// 节点值比较函数
int (*match)(void *ptr, void *key);
// 链表包含的节点数量
unsigned int len;
} list;

代码地址是这个https://github.com/antirez/redis/blob/2.2/src/adlist.h
可以看下节点是由listNode承载的,包括值和一个指向前节点跟一个指向后一节点的两个指针,然后值是 void 指针类型,所以可以承载不同类型的值
然后是 list结构用来承载一个链表,包含了表头,和表尾,复制函数,释放函数和比较函数,还有链表长度,因为包含了前两个节点,找到表尾节点跟表头都是 O(1)的时间复杂度,还有节点数量,其实这个跟 SDS 是同一个做法,就是空间换时间,这也是写代码里比较常见的做法,以此让一些高频的操作提速。

字典

字典也是个常用的数据结构,其实只是叫法不同,数据结构中叫 hash 散列,Java 中叫 Map,PHP 中是数组 array,Python 中也叫字典 dict,因为纯 C 语言本身不带这些数据结构,所以这也是个痛并快乐着的过程,享受 C 语言的高性能的同时也要接受它只提供了语言的基本功能的现实,各种轮子都需要自己造,redis 同样实现了自己的字典
下面来看看代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
typedef struct dictEntry {
void *key;
void *val;
struct dictEntry *next;
} dictEntry;

typedef struct dictType {
unsigned int (*hashFunction)(const void *key);
void *(*keyDup)(void *privdata, const void *key);
void *(*valDup)(void *privdata, const void *obj);
int (*keyCompare)(void *privdata, const void *key1, const void *key2);
void (*keyDestructor)(void *privdata, void *key);
void (*valDestructor)(void *privdata, void *obj);
} dictType;

/* This is our hash table structure. Every dictionary has two of this as we
* implement incremental rehashing, for the old to the new table. */
typedef struct dictht {
dictEntry **table;
unsigned long size;
unsigned long sizemask;
unsigned long used;
} dictht;

typedef struct dict {
dictType *type;
void *privdata;
dictht ht[2];
int rehashidx; /* rehashing not in progress if rehashidx == -1 */
int iterators; /* number of iterators currently running */
} dict;

看了下这个 2.2 版本的代码跟最新版的其实也差的不是很多,所以还是照旧用老代码,可以看到上面四个结构体中,其实只有三个是存储数据用的,dictType 是用来放操作函数的,那么三个存放数据的结构体分别是干嘛的,这时候感觉需要一个图来说明比较好,稍等,我去画个图~

这个图看着应该比较清楚这些都是用来干嘛的了,dict 是我们的主体结构,它有一个指向 dictType 的指针,这里面包含了字典的操作函数,然后是一个私有数据指针,接下来是一个 dictht 的数组,包含两个dictht,这个就是用来存数据的了,然后是 rehashidx 表示重哈希的状态,当是-1 的时候表示当前没有重哈希,iterators 表示正在遍历的迭代器的数量。
首先说说为啥需要有两个 dictht,这是因为字典 dict 这个数据结构随着数据量的增减,会需要在中途做扩容或者缩容操作,如果只有一个的话,对它进行扩容缩容时会影响正常的访问和修改操作,或者说保证正常查询,修改的正确性会比较复杂,并且因为需要高效利用空间,不能一下子申请一个非常大的空间来存很少的数据。当 dict 中 dictht 中的数据量超过 size 的时候负载就超过了 1,就需要进行扩容,这里的其实跟 Java 中的 HashMap 比较类似,超过一定的负载之后进行扩容。这里为啥 size 会超过 1 呢,可能有部分不了解这类结构的同学会比较奇怪,其实就是上图中画的,在数据结构中对于散列的冲突有几类解决方法,比如转换成链表,二次散列,找下个空槽等,这里就使用了链表法,或者说拉链法。当一个新元素通过 hashFunction 得出的 key 跟 sizemask 取模之后的值相同了,那就将其放在原来的节点之前,变成链表挂在数组 dictht.table下面,放在原有节点前是考虑到可能会优先访问。
忘了说明下 dictht 跟 dictEntry 的关系了,dictht 就是个哈希表,它里面是个dictEntry 的二维数组,而 dictEntry 是个包含了 key-value 结构之外还有一个 next 指针,因此可以将哈希冲突的以链表的形式保存下来。
在重点说下重哈希,可能同样写 Java 的同学对这个比较有感觉,跟 HashMap 一样,会以 2 的 N 次方进行扩容,那么扩容的方法就会比较简单,每个键重哈希要不就在原来这个槽,要不就在原来的槽加原 dictht.size 的位置;然后是重头戏,具体是怎么做扩容呢,其实这里就把第二个 ht 用上了,其实这两个hashtable 的具体作用有点类似于 jvm 中的两个 survival 区,但是又不全一样,因为 redis 在扩容的时候是采用的渐进式地重哈希,什么叫渐进式的呢,就是它不是像 jvm 那种标记复制的模式直接将一个 eden 区和原来的 survival 区存活的对象复制到另一个 survival 区,而是在每一次添加,删除,查找或者更新操作时,都会额外的帮忙搬运一部分的原 dictht 中的数据,这里会根据 rehashidx 的值来判断,如果是-1 表示并没有在重哈希中,如果是 0 表示开始重哈希了,然后rehashidx 还会随着每次的帮忙搬运往上加,但全部被搬运完成后 rehashidx 又变回了-1,又可以扯到Java 中的 Concurrent HashMap, 他在扩容的时候也使用了类似的操作。

这是个 Java 面试的高频问题,我也遇到过,以往都是觉得这类题没意思,网上一搜一大堆,也不愿意记,其实说回来,主要还是没静下心来好好去理解,今天无意中看到一个课程,基本帮我把一些疑惑的点讲清楚了,首先单例是啥意思,这个其实是有范围一说,比如我起了个Spring Boot应用,在这个应用范围内,我的常规 bean 是单例的,意味着 getBean 的时候其实永远只会拿到那一个对象,那要怎么来写一个单例呢,首先就是传说中的饿汉模式,也是最简单的

饿汉模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public class Singleton1 {
// 首先,将构造方法变成私有的
private Singleton1() {};
// 创建私有静态实例,这样第一次使用的时候就会进行创建
private static Singleton instance = new Singleton1();

// 使用这个对象都是通过这个 getInstance 来获取
public static Singleton1 getInstance() {
return instance;
}
// 瞎写一个静态方法。这里想说的是,如果我们只是要调用 Singleton.getDate(...),
// 本来是不想要生成 Singleton 实例的,不过没办法,已经生成了
public static Date getDate(String mode) {return new Date();}
}

上面借鉴了一些代码,其实这是最基本,也不会错的方法,但是正如其中getDate方法里说的问题,有时候并没有想那这个对象,但是因为我调用了这个类的静态方法,导致对象已经生成了,可能这也是饿汉模式名字的来由,不管三七二十一给你生成个单例就完事了,不管有没有用,但是这种个人觉得也没啥大问题,如果是面试的话最好说出来它的缺点

饱汉模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public class Singleton2 {
// 首先,也是先堵死 new Singleton() 这条路,将构造方法变成私有
private Singleton2() {}
// 和饿汉模式相比,这边不需要先实例化出来,注意这里的 volatile,它是必须的
private static volatile Singleton2 instance = null;

private int m = 9;

public static Singleton getInstance() {
if (instance == null) {
// 加锁
synchronized (Singleton2.class) {
// 这一次判断也是必须的,不然会有并发问题
if (instance == null) {
instance = new Singleton2();
}
}
}
return instance;
}
}

这里容易错的有三点,理解了其实就比较好记了

第一点,为啥不在 getInstance 上整个代码块加 synchronized,这个其实比较容易理解,就是锁的力度太大,性能太差了,这点其实也要去理解,可以举个夸张的例子,比如我一个电商的服务,如果为了避免一个人的订单出现问题,是不是可以从请求入口就把他锁住,到请求结束释放,那么里面做的事情都有保障,然而这显然不可能,因为我们想要这种竞态条件抢占资源的时间尽量减少,防止其他线程等待。
第二点,为啥synchronized之已经检查了 instance == null,还要在里面再检查一次,这个有个术语,叫 double check lock,但是为啥要这么做呢,其实很简单,想象当有两个线程,都过了第一步为空判断,这个时候只有一个线程能拿到这个锁,另一个线程就等待了,如果不再判断一次,那么第一个线程新建完对象释放锁之后,第二个线程又能拿到锁,再去创建一个对象。
第三点,为啥要volatile关键字,原先对它的理解是它修饰的变量在 JMM 中能及时将变量值写到主存中,但是它还有个很重要的作用,就是防止指令重排序,instance = new Singleton();这行代码其实在底层是分成三条指令执行的,第一条是在堆上申请了一块内存放这个对象,但是对象的字段啥的都还是默认值,第二条是设置对象的值,比如上面的 m 是 9,然后第三条是将这个对象和虚拟机栈上的指针建立引用关联,那么如果我不用volatile关键字,这三条指令就有可能出现重排,比如变成了 1-3-2 这种顺序,当执行完第二步时,有个线程来访问这个对象了,先判断是不是空,发现不是空的,就拿去直接用了,是不是就出现问题了,所以这个volatile也是不可缺少的

嵌套类

1
2
3
4
5
6
7
8
9
10
11
public class Singleton3 {

private Singleton3() {}
// 主要是使用了 嵌套类可以访问外部类的静态属性和静态方法 的特性
private static class Holder {
private static Singleton3 instance = new Singleton3();
}
public static Singleton3 getInstance() {
return Holder.instance;
}
}

这个我个人感觉是饿汉模式的升级版,可以在调用getInstance的时候去实例化对象,也是比较推荐的

枚举单例

1
2
3
4
5
6
7
public enum Singleton {
INSTANCE;

public void doSomething(){
//todo doSomething
}
}

枚举很特殊,它在类加载的时候会初始化里面的所有的实例,而且 JVM 保证了它们不会再被实例化,所以它天生就是单例的。

看完了村上春树的《1Q84》,这应该是第五本看的他的书了,继 跑步,挪威的森林,刺杀骑士团长,海边的卡夫卡之后,不是其中最长的,好像是海边的卡夫卡还是刺杀骑士团长比较长一点,都是在微信读书上看的,比较方便,最开始在上面看的是高晓松的《鱼羊野史》,不知道为啥取这个名字,但是还是满吸引我的,不过由于去年的种种,没有很多心思把它看完,而且本身的组织形式就是比较松散的,看到哪算哪,其实一些野史部分是我比较喜欢,有些谈到人物的就不太有兴趣,而且类似于大祥哥吃的东西,反正都是哇,怎么这么好吃,嗯,太爱(niu)你(bi)了,高晓松就是这个人是我最喜欢的 xxx 家,我也没去细究过他有没有说重复过,反正是不太爱,后来因为这书还一度对战争史有了浓厚的兴趣,然而事实告诉我,大部头的战争史,其实正史我是真的啃不下去,我可能只对其中 10%的内容感兴趣,不过终于也在今年把它看完了,好像高晓松的晓说也最终季了,貌似其中讲朝鲜战争的还被和谐了,看样子是说出了一些故事(truth)。

本来只是想把 《1Q84》的读后感写下,现在觉得还是把这篇当成我今年的读书总结吧,不过先从《1Q84》说起。

严格来讲,这不是很书面化的读后感,可能我想写的也只是像聊天一样的说下我读过的书,包括的技术博客其实也是类似的,以后或许会转变,但是目前水平如此吧,写多了可能会变好,也可能不会。

开始正文吧,这书有点类似于海边的卡夫卡,一开始是通过两条故事线,穿插着叙述,一条是青豆的,不算是个职业杀手的女杀手,要去解决一个经常家暴的斯文败类,穿着描述得比较性感吧,杀人方式是通过比较长的细针,从脖子后面一个精巧的位置插入,可以造成是未知原因死亡的假象,可能会推断成心梗之类的,这里有个前置的细节,就是青豆是乘坐一辆很高级的出租车,内饰什么的都非常有质感,有点不像一辆出租车,然后车里放了一首比较小众的歌,雅纳切克的《小交响曲》,但是青豆知道它,这跟后面的情节也有些许关系,这是女主人公青豆的出场;相应的男主的出场印象不是太深刻,男主叫天吾,是个不知名的作家,跟一个叫小松的编辑有比较好的关系,虽然天吾还没有拿到比较有分量的奖项,但是小松很看好他,也让他帮忙审校一个新作家奖的投稿文章,虽然天吾自身还没获得过这个奖,天吾还有个正式工作,是当数学老师,天吾在学生时代是个数学天才,但后面有对文学产生了兴趣,文学还不足以养活自己,靠着教课还是能保持温饱;

接下来是正式故事的起点了,就是小松收到了一部小说投稿,名叫《空气蛹》,是个叫深绘里的女孩子投的稿,小松对他赋予了很高的评价,这里好像记岔了,好像是天吾对这部小说很有好感,但是小松比较怀疑,然后小松看了之后也有了浓厚的兴趣,这里就是开端了,小松想让天吾来重写润色这部《空气蛹》,因为故事本身很有分量,但是描写手法叙事方式等都很拙劣,而天吾正好擅长这个,小松对天吾的评价是,描写技巧无可挑剔,就是故事主体的火花还没际遇迸发,需要一个导火索,这个就可以类比我们程序员,很多比较初中级的程序员主要擅长在原来的代码上修修改改或者给他分配个小功能,比较高级的程序员就需要能做一些项目的架构设计,核心的技术方案设计,以前我也觉得写文档这个比较无聊,但是当一个项目真的比较庞大,复杂的时候,整体和核心部分的架构设计和方案还是需要有文档沉淀的,不然别人不知道没法接受,自己过段时间也会忘记。

对于小松的这个建议,他的初衷是想搅一搅这个死气沉沉套路颇深的文坛,因为本身《空气蛹》这部小说的内容很吸引人,小松想通过天吾的润色补充让这部小说冲击新人奖,有种恶作剧的意图,天吾对此表示很多担心和顾虑,小松的这个建议其实也是一种文学作假,有两方面的担心,一方面是原作者深绘里是否同意如此操作,一方面是外界如果发现了这个事实会有什么样的后果,但是小松表示不用担心,前一步由小松牵线,让天吾跟原作者深绘里当面沟通这个代写是否被允许,结果当然是被允许了,这里有了对深绘里的初步描写,按我的理解是比较仙的感觉,然后语言沟通有些吃力,或者说有她自己的特色,当面沟通时貌似是让深绘里回去再考虑下,然后后面再由天吾去深绘里寄宿的戎野老师家沟通具体的细节。

2019年12月18日23:37:19 更新
去到戎野老师家之后,天吾知道了关于深绘里的一些事情,深绘里的父亲与戎野老师应该是老友,深绘里的父亲在当初成立了一个叫”先驱”的公社,一个独立运行的社会组织,以运营农场作为物资来源,追求更为松散的共同体,即不过分激进地公有制,进行松散的共同生活,承认私有财产,简而言之就是这样一个能稳定存活下来的独立社会组织,但是随着稳定运行,内部的激进派和稳健派开始出现分歧,不可磨合,后来两派就分裂了,深绘里的父亲,深田保留在了稳健派,但是此时其实深田保内心是矛盾的,以为一开始其实是他倡导的独立革命才组织起了这群人,然而现在他又认清了现实社会已经不太相信能通过革命来独立的可能性,后来激进派便开始越加封闭,而且进行军事训练和思想教育,而后这个先驱的激进派别便有了新的名字”黎明”,深绘里也是在此时从先驱逃离来投靠戎野老师
暂时先写到这,未完待续~

今天看了一下 redis 分布式锁 redlock 的实现,简单记录下,

加锁

原先我对 redis 锁的概念就是加锁使用 setnx,解锁使用 lua 脚本,但是 setnx 具体是啥,lua 脚本是啥不是很清楚
首先简单思考下这个问题,首先为啥不是先 get 一下 key 存不存在,然后再 set 一个 key value,因为加锁这个操作我们是要保证两点,一个是不能中途被打断,也就是说要原子性,如果是先 get 一下 key,如果不存在再 set 值的话,那就不是原子操作了;第二个是可不可以直接 set 值呢,显然不行,锁要保证唯一性,有且只能有一个线程或者其他应用单位获得该锁,正好 setnx 给了我们这种原子命令

然后是 setnx 的键和值分别是啥,键比较容易想到是要锁住的资源,比如 user_id, 这里有个我自己之前比较容易陷进去的误区,但是这个误区后
面再说,这里其实是把user_id 作为要锁住的资源,在我获得锁的时候别的线程不允许操作,以此保证业务的正确性,不会被多个线程同时修改,确定了键,再来看看值是啥,其实原先我认为值是啥都没关系,我只要锁住了,光键就够我用了,但是考虑下多个线程的问题,如果我这个线程加了锁,然后我因为 gc 停顿等原因卡死了,这个时候redis 的锁或者说就是 redis 的缓存已经过期了,这时候另一个线程获得锁成功,然后我这个线程又活过来了,然后我就仍然认为我拿着锁,我去对数据进行修改或者释放锁,是不是就出现问题了,所以是不是我们还需要一个东西来区分这个锁是哪个线程加的,所以我们可以将值设置成为一个线程独有识别的值,至少在相对长的一段时间内不会重复。

上面其实还有两个问题,一个是当 gc 超时时,我这个线程如何知道我手里的锁已经过期了,一种方法是我在加好锁之后就维护一个超时时间,这里其实还有个问题,不过跟第二个问题相关,就一起说了,就是设置超时时间,有些对于不是锁的 redis 缓存操作可以是先设置好值,然后在设置过期时间,那么这就又有上面说到的不是原子性的问题,那么就需要在同一条指令里把超时时间也设置了,幸好 redis 提供了这种支持

1
SET resource_name my_random_value NX PX 30000

这里借鉴一下解释下,resource_name就是 key,代表要锁住的东西,my_random_value就是识别我这个线程的,NX代表只有在不存在的时候才设置,然后PX 30000表示超时时间是 30秒自动过期

PS:记录下我原先有的一个误区,是不是要用 key 来区分加锁的线程,这样只有一个用处,就是自身线程可以识别是否是自己加的锁,但是最大的问题是别的线程不知道,其实这个用户的出发点是我在担心前面提过的一个问题,就是当 gc 停顿后,我要去判断当前的这个锁是否是我加的,还有就是当释放锁的时候,如果保证不会错误释放了其他线程加的锁,但是这样附带很多其他问题,最大的就是其他线程怎么知道能不能加这个锁。

解锁

当线程在锁过期之前就处理完了业务逻辑,那就可以提前释放这个锁,那么提前释放要怎么操作,直接del key显然是不行的,因为这样就是我前面想用线程随机值加资源名作为锁的初衷,我不能去释放别的线程加的锁,那么我要怎么办呢,先 get 一下看是不是我的?那又变成非原子的操作了,幸好redis 也考虑到了这个问题,给了lua 脚本来操作这种

1
2
3
4
5
if redis.call("get",KEYS[1]) == ARGV[1] then
return redis.call("del",KEYS[1])
else
return 0
end

这里的KEYS[1]就是前面加锁的resource_name,ARGV[1]就是线程的随机值my_random_value

多节点

前面说的其实是单节点 redis 作为分布式锁的情况,那么当我们的 redis 有多节点的情况呢,如果多节点下处于加锁或者解锁或者锁有效情况下
redis 的某个节点宕掉了怎么办,这里就有一些需要思考的地方,是否单独搞一个单节点的 redis作为分布式锁专用的,但是如果这个单节点的挂了呢,还有就是成本问题,所以我们需要一个多节点的分布式锁方案
这里就引出了开头说到的redlock,这个可是 redis的作者写的, 他的加锁过程是分以下几步去做这个事情

  • 获取当前时间(毫秒数)。
  • 按顺序依次向N个Redis节点执行获取锁的操作。这个获取操作跟前面基于单Redis节点的获取锁的过程相同,包含随机字符串my_random_value,也包含过期时间(比如PX 30000,即锁的有效时间)。为了保证在某个Redis节点不可用的时候算法能够继续运行,这个获取锁的操作还有一个超时时间(time out),它要远小于锁的有效时间(几十毫秒量级)。客户端在向某个Redis节点获取锁失败以后,应该立即尝试下一个Redis节点。这里的失败,应该包含任何类型的失败,比如该Redis节点不可用,或者该Redis节点上的锁已经被其它客户端持有(注:Redlock原文中这里只提到了Redis节点不可用的情况,但也应该包含其它的失败情况)。
  • 计算整个获取锁的过程总共消耗了多长时间,计算方法是用当前时间减去第1步记录的时间。如果客户端从大多数Redis节点(>= N/2+1)成功获取到了锁,并且获取锁总共消耗的时间没有超过锁的有效时间(lock validity time),那么这时客户端才认为最终获取锁成功;否则,认为最终获取锁失败。
  • 如果最终获取锁成功了,那么这个锁的有效时间应该重新计算,它等于最初的锁的有效时间减去第3步计算出来的获取锁消耗的时间。
  • 如果最终获取锁失败了(可能由于获取到锁的Redis节点个数少于N/2+1,或者整个获取锁的过程消耗的时间超过了锁的最初有效时间),那么客户端应该立即向所有Redis节点发起释放锁的操作(即前面介绍的Redis Lua脚本)。
    释放锁的过程比较简单:客户端向所有Redis节点发起释放锁的操作,不管这些节点当时在获取锁的时候成功与否。这里为什么要向所有的节点发送释放锁的操作呢,这里是因为有部分的节点的失败原因可能是加锁时阻塞,加锁成功的结果没有及时返回,所以为了防止这种情况还是需要向所有发起这个释放锁的操作。
    初步记录就先到这。

对 Java 的 gc 实现比较感兴趣,原先一般都是看周志明的书,但其实并没有讲具体的 gc 源码,而是把整个思路和流程讲解了一下
特别是 G1 的具体实现
一般对 G1 的理解其实就是把原先整块的新生代老年代分成了以 region 为单位的小块内存,简而言之,就是原先对新生代老年代的收集会涉及到整个代的堆内存空间,而G1 把它变成了更细致的小块内存
这带来了一个很明显的好处和一个很明显的坏处,好处是内存收集可以更灵活,耗时会变短,但整个收集的处理复杂度就变高了
目前看了一点点关于 G1 收集的预期时间相关的代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
HeapWord* G1CollectedHeap::do_collection_pause(size_t word_size,
uint gc_count_before,
bool* succeeded,
GCCause::Cause gc_cause) {
assert_heap_not_locked_and_not_at_safepoint();
VM_G1CollectForAllocation op(word_size,
gc_count_before,
gc_cause,
false, /* should_initiate_conc_mark */
g1_policy()->max_pause_time_ms());
VMThread::execute(&op);

HeapWord* result = op.result();
bool ret_succeeded = op.prologue_succeeded() && op.pause_succeeded();
assert(result == NULL || ret_succeeded,
"the result should be NULL if the VM did not succeed");
*succeeded = ret_succeeded;

assert_heap_not_locked();
return result;
}

这里就是收集时需要停顿的,其中VMThread::execute(&op);是具体执行的,真正执行的是VM_G1CollectForAllocation::doit方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
void VM_G1CollectForAllocation::doit() {
G1CollectedHeap* g1h = G1CollectedHeap::heap();
assert(!_should_initiate_conc_mark || g1h->should_do_concurrent_full_gc(_gc_cause),
"only a GC locker, a System.gc(), stats update, whitebox, or a hum allocation induced GC should start a cycle");

if (_word_size > 0) {
// An allocation has been requested. So, try to do that first.
_result = g1h->attempt_allocation_at_safepoint(_word_size,
false /* expect_null_cur_alloc_region */);
if (_result != NULL) {
// If we can successfully allocate before we actually do the
// pause then we will consider this pause successful.
_pause_succeeded = true;
return;
}
}

GCCauseSetter x(g1h, _gc_cause);
if (_should_initiate_conc_mark) {
// It's safer to read old_marking_cycles_completed() here, given
// that noone else will be updating it concurrently. Since we'll
// only need it if we're initiating a marking cycle, no point in
// setting it earlier.
_old_marking_cycles_completed_before = g1h->old_marking_cycles_completed();

// At this point we are supposed to start a concurrent cycle. We
// will do so if one is not already in progress.
bool res = g1h->g1_policy()->force_initial_mark_if_outside_cycle(_gc_cause);

// The above routine returns true if we were able to force the
// next GC pause to be an initial mark; it returns false if a
// marking cycle is already in progress.
//
// If a marking cycle is already in progress just return and skip the
// pause below - if the reason for requesting this initial mark pause
// was due to a System.gc() then the requesting thread should block in
// doit_epilogue() until the marking cycle is complete.
//
// If this initial mark pause was requested as part of a humongous
// allocation then we know that the marking cycle must just have
// been started by another thread (possibly also allocating a humongous
// object) as there was no active marking cycle when the requesting
// thread checked before calling collect() in
// attempt_allocation_humongous(). Retrying the GC, in this case,
// will cause the requesting thread to spin inside collect() until the
// just started marking cycle is complete - which may be a while. So
// we do NOT retry the GC.
if (!res) {
assert(_word_size == 0, "Concurrent Full GC/Humongous Object IM shouldn't be allocating");
if (_gc_cause != GCCause::_g1_humongous_allocation) {
_should_retry_gc = true;
}
return;
}
}

// Try a partial collection of some kind.
_pause_succeeded = g1h->do_collection_pause_at_safepoint(_target_pause_time_ms);

if (_pause_succeeded) {
if (_word_size > 0) {
// An allocation had been requested. Do it, eventually trying a stronger
// kind of GC.
_result = g1h->satisfy_failed_allocation(_word_size, &_pause_succeeded);
} else {
bool should_upgrade_to_full = !g1h->should_do_concurrent_full_gc(_gc_cause) &&
!g1h->has_regions_left_for_allocation();
if (should_upgrade_to_full) {
// There has been a request to perform a GC to free some space. We have no
// information on how much memory has been asked for. In case there are
// absolutely no regions left to allocate into, do a maximally compacting full GC.
log_info(gc, ergo)("Attempting maximally compacting collection");
_pause_succeeded = g1h->do_full_collection(false, /* explicit gc */
true /* clear_all_soft_refs */);
}
}
guarantee(_pause_succeeded, "Elevated collections during the safepoint must always succeed.");
} else {
assert(_result == NULL, "invariant");
// The only reason for the pause to not be successful is that, the GC locker is
// active (or has become active since the prologue was executed). In this case
// we should retry the pause after waiting for the GC locker to become inactive.
_should_retry_gc = true;
}
}

这里可以看到核心的是G1CollectedHeap::do_collection_pause_at_safepoint这个方法,它带上了目标暂停时间的值

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
G1CollectedHeap::do_collection_pause_at_safepoint(double target_pause_time_ms) {
assert_at_safepoint_on_vm_thread();
guarantee(!is_gc_active(), "collection is not reentrant");

if (GCLocker::check_active_before_gc()) {
return false;
}

_gc_timer_stw->register_gc_start();

GCIdMark gc_id_mark;
_gc_tracer_stw->report_gc_start(gc_cause(), _gc_timer_stw->gc_start());

SvcGCMarker sgcm(SvcGCMarker::MINOR);
ResourceMark rm;

g1_policy()->note_gc_start();

wait_for_root_region_scanning();

print_heap_before_gc();
print_heap_regions();
trace_heap_before_gc(_gc_tracer_stw);

_verifier->verify_region_sets_optional();
_verifier->verify_dirty_young_regions();

// We should not be doing initial mark unless the conc mark thread is running
if (!_cm_thread->should_terminate()) {
// This call will decide whether this pause is an initial-mark
// pause. If it is, in_initial_mark_gc() will return true
// for the duration of this pause.
g1_policy()->decide_on_conc_mark_initiation();
}

// We do not allow initial-mark to be piggy-backed on a mixed GC.
assert(!collector_state()->in_initial_mark_gc() ||
collector_state()->in_young_only_phase(), "sanity");

// We also do not allow mixed GCs during marking.
assert(!collector_state()->mark_or_rebuild_in_progress() || collector_state()->in_young_only_phase(), "sanity");

// Record whether this pause is an initial mark. When the current
// thread has completed its logging output and it's safe to signal
// the CM thread, the flag's value in the policy has been reset.
bool should_start_conc_mark = collector_state()->in_initial_mark_gc();

// Inner scope for scope based logging, timers, and stats collection
{
EvacuationInfo evacuation_info;

if (collector_state()->in_initial_mark_gc()) {
// We are about to start a marking cycle, so we increment the
// full collection counter.
increment_old_marking_cycles_started();
_cm->gc_tracer_cm()->set_gc_cause(gc_cause());
}

_gc_tracer_stw->report_yc_type(collector_state()->yc_type());

GCTraceCPUTime tcpu;

G1HeapVerifier::G1VerifyType verify_type;
FormatBuffer<> gc_string("Pause Young ");
if (collector_state()->in_initial_mark_gc()) {
gc_string.append("(Concurrent Start)");
verify_type = G1HeapVerifier::G1VerifyConcurrentStart;
} else if (collector_state()->in_young_only_phase()) {
if (collector_state()->in_young_gc_before_mixed()) {
gc_string.append("(Prepare Mixed)");
} else {
gc_string.append("(Normal)");
}
verify_type = G1HeapVerifier::G1VerifyYoungNormal;
} else {
gc_string.append("(Mixed)");
verify_type = G1HeapVerifier::G1VerifyMixed;
}
GCTraceTime(Info, gc) tm(gc_string, NULL, gc_cause(), true);

uint active_workers = AdaptiveSizePolicy::calc_active_workers(workers()->total_workers(),
workers()->active_workers(),
Threads::number_of_non_daemon_threads());
active_workers = workers()->update_active_workers(active_workers);
log_info(gc,task)("Using %u workers of %u for evacuation", active_workers, workers()->total_workers());

TraceCollectorStats tcs(g1mm()->incremental_collection_counters());
TraceMemoryManagerStats tms(&_memory_manager, gc_cause(),
collector_state()->yc_type() == Mixed /* allMemoryPoolsAffected */);

G1HeapTransition heap_transition(this);
size_t heap_used_bytes_before_gc = used();

// Don't dynamically change the number of GC threads this early. A value of
// 0 is used to indicate serial work. When parallel work is done,
// it will be set.

{ // Call to jvmpi::post_class_unload_events must occur outside of active GC
IsGCActiveMark x;

gc_prologue(false);

if (VerifyRememberedSets) {
log_info(gc, verify)("[Verifying RemSets before GC]");
VerifyRegionRemSetClosure v_cl;
heap_region_iterate(&v_cl);
}

_verifier->verify_before_gc(verify_type);

_verifier->check_bitmaps("GC Start");

#if COMPILER2_OR_JVMCI
DerivedPointerTable::clear();
#endif

// Please see comment in g1CollectedHeap.hpp and
// G1CollectedHeap::ref_processing_init() to see how
// reference processing currently works in G1.

// Enable discovery in the STW reference processor
_ref_processor_stw->enable_discovery();

{
// We want to temporarily turn off discovery by the
// CM ref processor, if necessary, and turn it back on
// on again later if we do. Using a scoped
// NoRefDiscovery object will do this.
NoRefDiscovery no_cm_discovery(_ref_processor_cm);

// Forget the current alloc region (we might even choose it to be part
// of the collection set!).
_allocator->release_mutator_alloc_region();

// This timing is only used by the ergonomics to handle our pause target.
// It is unclear why this should not include the full pause. We will
// investigate this in CR 7178365.
//
// Preserving the old comment here if that helps the investigation:
//
// The elapsed time induced by the start time below deliberately elides
// the possible verification above.
double sample_start_time_sec = os::elapsedTime();

g1_policy()->record_collection_pause_start(sample_start_time_sec);

if (collector_state()->in_initial_mark_gc()) {
concurrent_mark()->pre_initial_mark();
}

g1_policy()->finalize_collection_set(target_pause_time_ms, &_survivor);

evacuation_info.set_collectionset_regions(collection_set()->region_length());

// Make sure the remembered sets are up to date. This needs to be
// done before register_humongous_regions_with_cset(), because the
// remembered sets are used there to choose eager reclaim candidates.
// If the remembered sets are not up to date we might miss some
// entries that need to be handled.
g1_rem_set()->cleanupHRRS();

register_humongous_regions_with_cset();

assert(_verifier->check_cset_fast_test(), "Inconsistency in the InCSetState table.");

// We call this after finalize_cset() to
// ensure that the CSet has been finalized.
_cm->verify_no_cset_oops();

if (_hr_printer.is_active()) {
G1PrintCollectionSetClosure cl(&_hr_printer);
_collection_set.iterate(&cl);
}

// Initialize the GC alloc regions.
_allocator->init_gc_alloc_regions(evacuation_info);

G1ParScanThreadStateSet per_thread_states(this, workers()->active_workers(), collection_set()->young_region_length());
pre_evacuate_collection_set();

// Actually do the work...
evacuate_collection_set(&per_thread_states);

post_evacuate_collection_set(evacuation_info, &per_thread_states);

const size_t* surviving_young_words = per_thread_states.surviving_young_words();
free_collection_set(&_collection_set, evacuation_info, surviving_young_words);

eagerly_reclaim_humongous_regions();

record_obj_copy_mem_stats();
_survivor_evac_stats.adjust_desired_plab_sz();
_old_evac_stats.adjust_desired_plab_sz();

double start = os::elapsedTime();
start_new_collection_set();
g1_policy()->phase_times()->record_start_new_cset_time_ms((os::elapsedTime() - start) * 1000.0);

if (evacuation_failed()) {
set_used(recalculate_used());
if (_archive_allocator != NULL) {
_archive_allocator->clear_used();
}
for (uint i = 0; i < ParallelGCThreads; i++) {
if (_evacuation_failed_info_array[i].has_failed()) {
_gc_tracer_stw->report_evacuation_failed(_evacuation_failed_info_array[i]);
}
}
} else {
// The "used" of the the collection set have already been subtracted
// when they were freed. Add in the bytes evacuated.
increase_used(g1_policy()->bytes_copied_during_gc());
}

if (collector_state()->in_initial_mark_gc()) {
// We have to do this before we notify the CM threads that
// they can start working to make sure that all the
// appropriate initialization is done on the CM object.
concurrent_mark()->post_initial_mark();
// Note that we don't actually trigger the CM thread at
// this point. We do that later when we're sure that
// the current thread has completed its logging output.
}

allocate_dummy_regions();

_allocator->init_mutator_alloc_region();

{
size_t expand_bytes = _heap_sizing_policy->expansion_amount();
if (expand_bytes > 0) {
size_t bytes_before = capacity();
// No need for an ergo logging here,
// expansion_amount() does this when it returns a value > 0.
double expand_ms;
if (!expand(expand_bytes, _workers, &expand_ms)) {
// We failed to expand the heap. Cannot do anything about it.
}
g1_policy()->phase_times()->record_expand_heap_time(expand_ms);
}
}

// We redo the verification but now wrt to the new CSet which
// has just got initialized after the previous CSet was freed.
_cm->verify_no_cset_oops();

// This timing is only used by the ergonomics to handle our pause target.
// It is unclear why this should not include the full pause. We will
// investigate this in CR 7178365.
double sample_end_time_sec = os::elapsedTime();
double pause_time_ms = (sample_end_time_sec - sample_start_time_sec) * MILLIUNITS;
size_t total_cards_scanned = g1_policy()->phase_times()->sum_thread_work_items(G1GCPhaseTimes::ScanRS, G1GCPhaseTimes::ScanRSScannedCards);
g1_policy()->record_collection_pause_end(pause_time_ms, total_cards_scanned, heap_used_bytes_before_gc);

evacuation_info.set_collectionset_used_before(collection_set()->bytes_used_before());
evacuation_info.set_bytes_copied(g1_policy()->bytes_copied_during_gc());

if (VerifyRememberedSets) {
log_info(gc, verify)("[Verifying RemSets after GC]");
VerifyRegionRemSetClosure v_cl;
heap_region_iterate(&v_cl);
}

_verifier->verify_after_gc(verify_type);
_verifier->check_bitmaps("GC End");

assert(!_ref_processor_stw->discovery_enabled(), "Postcondition");
_ref_processor_stw->verify_no_references_recorded();

// CM reference discovery will be re-enabled if necessary.
}

#ifdef TRACESPINNING
ParallelTaskTerminator::print_termination_counts();
#endif

gc_epilogue(false);
}

// Print the remainder of the GC log output.
if (evacuation_failed()) {
log_info(gc)("To-space exhausted");
}

g1_policy()->print_phases();
heap_transition.print();

// It is not yet to safe to tell the concurrent mark to
// start as we have some optional output below. We don't want the
// output from the concurrent mark thread interfering with this
// logging output either.

_hrm.verify_optional();
_verifier->verify_region_sets_optional();

TASKQUEUE_STATS_ONLY(print_taskqueue_stats());
TASKQUEUE_STATS_ONLY(reset_taskqueue_stats());

print_heap_after_gc();
print_heap_regions();
trace_heap_after_gc(_gc_tracer_stw);

// We must call G1MonitoringSupport::update_sizes() in the same scoping level
// as an active TraceMemoryManagerStats object (i.e. before the destructor for the
// TraceMemoryManagerStats is called) so that the G1 memory pools are updated
// before any GC notifications are raised.
g1mm()->update_sizes();

_gc_tracer_stw->report_evacuation_info(&evacuation_info);
_gc_tracer_stw->report_tenuring_threshold(_g1_policy->tenuring_threshold());
_gc_timer_stw->register_gc_end();
_gc_tracer_stw->report_gc_end(_gc_timer_stw->gc_end(), _gc_timer_stw->time_partitions());
}
// It should now be safe to tell the concurrent mark thread to start
// without its logging output interfering with the logging output
// that came from the pause.

if (should_start_conc_mark) {
// CAUTION: after the doConcurrentMark() call below,
// the concurrent marking thread(s) could be running
// concurrently with us. Make sure that anything after
// this point does not assume that we are the only GC thread
// running. Note: of course, the actual marking work will
// not start until the safepoint itself is released in
// SuspendibleThreadSet::desynchronize().
do_concurrent_mark();
}

return true;
}

往下走就是这一步G1Policy::finalize_collection_set,去处理新生代和老年代

1
2
3
4
void G1Policy::finalize_collection_set(double target_pause_time_ms, G1SurvivorRegions* survivor) {
double time_remaining_ms = _collection_set->finalize_young_part(target_pause_time_ms, survivor);
_collection_set->finalize_old_part(time_remaining_ms);
}

这里分别调用了两个方法,可以看到剩余时间是往下传的,来看一下具体的方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
double G1CollectionSet::finalize_young_part(double target_pause_time_ms, G1SurvivorRegions* survivors) {
double young_start_time_sec = os::elapsedTime();

finalize_incremental_building();

guarantee(target_pause_time_ms > 0.0,
"target_pause_time_ms = %1.6lf should be positive", target_pause_time_ms);

size_t pending_cards = _policy->pending_cards();
double base_time_ms = _policy->predict_base_elapsed_time_ms(pending_cards);
double time_remaining_ms = MAX2(target_pause_time_ms - base_time_ms, 0.0);

log_trace(gc, ergo, cset)("Start choosing CSet. pending cards: " SIZE_FORMAT " predicted base time: %1.2fms remaining time: %1.2fms target pause time: %1.2fms",
pending_cards, base_time_ms, time_remaining_ms, target_pause_time_ms);

// The young list is laid with the survivor regions from the previous
// pause are appended to the RHS of the young list, i.e.
// [Newly Young Regions ++ Survivors from last pause].

uint survivor_region_length = survivors->length();
uint eden_region_length = _g1h->eden_regions_count();
init_region_lengths(eden_region_length, survivor_region_length);

verify_young_cset_indices();

// Clear the fields that point to the survivor list - they are all young now.
survivors->convert_to_eden();

_bytes_used_before = _inc_bytes_used_before;
time_remaining_ms = MAX2(time_remaining_ms - _inc_predicted_elapsed_time_ms, 0.0);

log_trace(gc, ergo, cset)("Add young regions to CSet. eden: %u regions, survivors: %u regions, predicted young region time: %1.2fms, target pause time: %1.2fms",
eden_region_length, survivor_region_length, _inc_predicted_elapsed_time_ms, target_pause_time_ms);

// The number of recorded young regions is the incremental
// collection set's current size
set_recorded_rs_lengths(_inc_recorded_rs_lengths);

double young_end_time_sec = os::elapsedTime();
phase_times()->record_young_cset_choice_time_ms((young_end_time_sec - young_start_time_sec) * 1000.0);

return time_remaining_ms;
}

下面是老年代的部分

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
void G1CollectionSet::finalize_old_part(double time_remaining_ms) {
double non_young_start_time_sec = os::elapsedTime();
double predicted_old_time_ms = 0.0;

if (collector_state()->in_mixed_phase()) {
cset_chooser()->verify();
const uint min_old_cset_length = _policy->calc_min_old_cset_length();
const uint max_old_cset_length = _policy->calc_max_old_cset_length();

uint expensive_region_num = 0;
bool check_time_remaining = _policy->adaptive_young_list_length();

HeapRegion* hr = cset_chooser()->peek();
while (hr != NULL) {
if (old_region_length() >= max_old_cset_length) {
// Added maximum number of old regions to the CSet.
log_debug(gc, ergo, cset)("Finish adding old regions to CSet (old CSet region num reached max). old %u regions, max %u regions",
old_region_length(), max_old_cset_length);
break;
}

// Stop adding regions if the remaining reclaimable space is
// not above G1HeapWastePercent.
size_t reclaimable_bytes = cset_chooser()->remaining_reclaimable_bytes();
double reclaimable_percent = _policy->reclaimable_bytes_percent(reclaimable_bytes);
double threshold = (double) G1HeapWastePercent;
if (reclaimable_percent <= threshold) {
// We've added enough old regions that the amount of uncollected
// reclaimable space is at or below the waste threshold. Stop
// adding old regions to the CSet.
log_debug(gc, ergo, cset)("Finish adding old regions to CSet (reclaimable percentage not over threshold). "
"old %u regions, max %u regions, reclaimable: " SIZE_FORMAT "B (%1.2f%%) threshold: " UINTX_FORMAT "%%",
old_region_length(), max_old_cset_length, reclaimable_bytes, reclaimable_percent, G1HeapWastePercent);
break;
}

double predicted_time_ms = predict_region_elapsed_time_ms(hr);
if (check_time_remaining) {
if (predicted_time_ms > time_remaining_ms) {
// Too expensive for the current CSet.

if (old_region_length() >= min_old_cset_length) {
// We have added the minimum number of old regions to the CSet,
// we are done with this CSet.
log_debug(gc, ergo, cset)("Finish adding old regions to CSet (predicted time is too high). "
"predicted time: %1.2fms, remaining time: %1.2fms old %u regions, min %u regions",
predicted_time_ms, time_remaining_ms, old_region_length(), min_old_cset_length);
break;
}

// We'll add it anyway given that we haven't reached the
// minimum number of old regions.
expensive_region_num += 1;
}
} else {
if (old_region_length() >= min_old_cset_length) {
// In the non-auto-tuning case, we'll finish adding regions
// to the CSet if we reach the minimum.

log_debug(gc, ergo, cset)("Finish adding old regions to CSet (old CSet region num reached min). old %u regions, min %u regions",
old_region_length(), min_old_cset_length);
break;
}
}

// We will add this region to the CSet.
time_remaining_ms = MAX2(time_remaining_ms - predicted_time_ms, 0.0);
predicted_old_time_ms += predicted_time_ms;
cset_chooser()->pop(); // already have region via peek()
_g1h->old_set_remove(hr);
add_old_region(hr);

hr = cset_chooser()->peek();
}
if (hr == NULL) {
log_debug(gc, ergo, cset)("Finish adding old regions to CSet (candidate old regions not available)");
}

if (expensive_region_num > 0) {
// We print the information once here at the end, predicated on
// whether we added any apparently expensive regions or not, to
// avoid generating output per region.
log_debug(gc, ergo, cset)("Added expensive regions to CSet (old CSet region num not reached min)."
"old: %u regions, expensive: %u regions, min: %u regions, remaining time: %1.2fms",
old_region_length(), expensive_region_num, min_old_cset_length, time_remaining_ms);
}

cset_chooser()->verify();
}

stop_incremental_building();

log_debug(gc, ergo, cset)("Finish choosing CSet. old: %u regions, predicted old region time: %1.2fms, time remaining: %1.2f",
old_region_length(), predicted_old_time_ms, time_remaining_ms);

double non_young_end_time_sec = os::elapsedTime();
phase_times()->record_non_young_cset_choice_time_ms((non_young_end_time_sec - non_young_start_time_sec) * 1000.0);

QuickSort::sort(_collection_set_regions, _collection_set_cur_length, compare_region_idx, true);
}

上面第三行是个判断,当前是否是 mixed 回收阶段,如果不是的话其实是没有老年代什么事的,所以可以看到代码基本是从这个 if 判断
if (collector_state()->in_mixed_phase()) {开始往下走的
先写到这,偏向于做笔记用,有错轻拍

最近看了大神的 AQS 的文章,之前总是断断续续地看一点,每次都知难而退,下次看又从头开始,昨天总算硬着头皮看完了第一部分
首先 AQS 只要有这些属性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 头结点,你直接把它当做 当前持有锁的线程 可能是最好理解的
private transient volatile Node head;

// 阻塞的尾节点,每个新的节点进来,都插入到最后,也就形成了一个链表
private transient volatile Node tail;

// 这个是最重要的,代表当前锁的状态,0代表没有被占用,大于 0 代表有线程持有当前锁
// 这个值可以大于 1,是因为锁可以重入,每次重入都加上 1
private volatile int state;

// 代表当前持有独占锁的线程,举个最重要的使用例子,因为锁可以重入
// reentrantLock.lock()可以嵌套调用多次,所以每次用这个来判断当前线程是否已经拥有了锁
// if (currentThread == getExclusiveOwnerThread()) {state++}
private transient Thread exclusiveOwnerThread; //继承自AbstractOwnableSynchronizer

大概了解了 aqs 底层的双向等待队列,
结构是这样的

每个 node 里面主要是的代码结构也比较简单

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
static final class Node {
// 标识节点当前在共享模式下
static final Node SHARED = new Node();
// 标识节点当前在独占模式下
static final Node EXCLUSIVE = null;

// ======== 下面的几个int常量是给waitStatus用的 ===========
/** waitStatus value to indicate thread has cancelled */
// 代码此线程取消了争抢这个锁
static final int CANCELLED = 1;
/** waitStatus value to indicate successor's thread needs unparking */
// 官方的描述是,其表示当前node的后继节点对应的线程需要被唤醒
static final int SIGNAL = -1;
/** waitStatus value to indicate thread is waiting on condition */
// 本文不分析condition,所以略过吧,下一篇文章会介绍这个
static final int CONDITION = -2;
/**
* waitStatus value to indicate the next acquireShared should
* unconditionally propagate
*/
// 同样的不分析,略过吧
static final int PROPAGATE = -3;
// =====================================================


// 取值为上面的1、-1、-2、-3,或者0(以后会讲到)
// 这么理解,暂时只需要知道如果这个值 大于0 代表此线程取消了等待,
// ps: 半天抢不到锁,不抢了,ReentrantLock是可以指定timeouot的。。。
volatile int waitStatus;
// 前驱节点的引用
volatile Node prev;
// 后继节点的引用
volatile Node next;
// 这个就是线程本尊
volatile Thread thread;

}

其实可以主要关注这个 waitStatus 因为这个是后面的节点给前面的节点设置的,等于-1 的时候代表后面有节点等待,需要去唤醒,
这里使用了一个变种的 CLH 队列实现,CLH 队列相关内容可以查看这篇 自旋锁、排队自旋锁、MCS锁、CLH锁