}

Unsafe与CAS

在ConcurrentHashMap中，随处可以看到Unsafe, 大量使用了Unsafe.compareAndSwapXXX的方法，这个方法是利用一个CAS算法实现无锁化的修改值的操作，他可以大大降低锁代理的性能消耗。

这个算法的基本思想就是不断地去比较当前内存中的变量值与你指定的一个变量值是否相等，如果相等，则接受你指定的修改的值，否则拒绝你的操作。

因为当前线程中的值已经不是最新的值，你的修改很可能会覆盖掉其他线程修改的结果。这一点与乐观锁，SVN的思想是比较类似的。

unsafe静态块

unsafe代码块控制了一些属性的修改工作，比如最常用的SIZECTL 。 在这一版本的concurrentHashMap中，大量应用来的CAS方法进行变量、属性的修改工作。 利用CAS进行无锁操作，可以大大提高性能。

// 各个字段属性的偏移量，可以通过偏移量+首地址获取到对应的数据对象

private static final sun.misc.Unsafe U;

private static final long SIZECTL;

private static final long TRANSFERINDEX;

private static final long BASECOUNT;

private static final long CELLSBUSY;

private static final long CELLVALUE;

private static final long ABASE;

private static final int ASHIFT;

static {

try {

U = sun.misc.Unsafe.getUnsafe();

Class<?> k = ConcurrentHashMap.class;

SIZECTL = U.objectFieldOffset

(k.getDeclaredField(“sizeCtl”));

TRANSFERINDEX = U.objectFieldOffset

(k.getDeclaredField(“transferIndex”));

BASECOUNT = U.objectFieldOffset

(k.getDeclaredField(“baseCount”));

CELLSBUSY = U.objectFieldOffset

(k.getDeclaredField(“cellsBusy”));

Class<?> ck = CounterCell.class;

CELLVALUE = U.objectFieldOffset

(ck.getDeclaredField(“value”));

Class<?> ak = Node[].class;

ABASE = U.arrayBaseOffset(ak);

int scale = U.arrayIndexScale(ak);

if ((scale & (scale - 1)) != 0)

throw new Error(“data type scale not a power of two”);

ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);

} catch (Exception e) {

throw new Error(e);

}

}

三个核心方法

ConcurrentHashMap定义了三个原子操作，用于对指定位置的节点进行操作。正是这些原子操作保证了ConcurrentHashMap的线程安全。

@SuppressWarnings(“unchecked”)

//获得在i位置上的Node节点

static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {

return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);

}

• 利用CAS算法设置i位置上的Node节点。之所以能实现并发是因为他指定了原来这个节点的值是多少

• 在CAS算法中，会比较内存中的值与你指定的这个值是否相等，如果相等才接受你的修改，否则拒绝你的修改

• 因此当前线程中的值并不是最新的值，这种修改可能会覆盖掉其他线程的修改结果 有点类似于SVN

static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,

Node<K,V> c, Node<K,V> v) {

return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);

}

//利用volatile方法设置节点位置的值

static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {

U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);

}

初始化方法initTable

• 对于ConcurrentHashMap来说，调用它的构造方法仅仅是设置一些参数。

• 整个table的初始化是在向ConcurrentHashMap中插入元素的时候发生的。

• 如调用put、computeIfAbsent、compute、merge等方法的时候，调用时机是检查table==null。

1. 初始化方法主要应用了关键属性sizeCtl 如果这个值 <0，表示其他线程正在进行初始化，就放弃这个操作。

2. 在这也可以看出ConcurrentHashMap的初始化只能由一个线程完成。

3. 如果获得了初始化权限，就用CAS方法将sizeCtl置为-1，防止其他线程进入。

初始化数组后，将sizeCtl的值改为0.75 * n，源码如下：

/**

• Initializes table, using the size recorded in sizeCtl.

*/

private final Node<K,V>[] initTable() {

Node<K,V>[] tab; int sc;

while ((tab = table) == null || tab.length == 0) {

//sizeCtl表示有其他线程正在进行初始化操作，把线程挂起。对于table的初始化工作，只能有一个线程在进行。

if ((sc = sizeCtl) < 0)

Thread.yield(); // lost initialization race; just spin

else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//利用CAS方法把sizectl的值置为-1 表示本线程正在进行初始化

try {

if ((tab = table) == null || tab.length == 0) {

int n = (sc > 0) ? sc : DEFAULT_CAPACITY;

@SuppressWarnings(“unchecked”)

Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];

table = tab = nt;

sc = n - (n >>> 2);//相当于0.75*n 设置一个扩容的阈值

}

} finally {

sizeCtl = sc;

}

break;

}

}

return tab;

}

扩容方法 transfer

• ConcurrentHashMap容量不足的时候，需要对table进行扩容。

• 这个方法的基本思想跟HashMap是很像的，但是由于它是支持并发扩容的，所以要复杂的多。

• 原因是它支持多线程进行扩容操作，而并没有加锁。

• 我想这样做的目的不仅仅是为了满足concurrent的要求，而是希望利用并发处理去减少扩容带来的时间影响。因为在扩容的时候，总是会涉及到从一个“数组”到另一个“数组”拷贝的操作，如果这个操作能够并发进行，那真真是极好的了。

整个扩容操作分为两个部分

• 第一部分是构建一个nextTable,它的容量是原来的两倍，这个操作是单线程完成的。这个单线程的保证是通过RESIZE_STAMP_SHIFT这个常量经过一次运算来保证的，这个地方在后面会有提到；

• 第二个部分就是将原来table中的元素复制到nextTable中，这里允许多线程进行操作。

先来看一下单线程是如何完成的：

• 它的大体思想就是遍历、复制的过程。首先根据运算得到需要遍历的次数i，然后利用tabAt方法获得i位置的元素：

• 如果这个位置为空，就在原table中的i位置放入forwardNode节点，这个也是触发并发扩容的关键点；

• 如果这个位置是Node节点（fh>=0），如果它是一个链表的头节点，就构造一个反序链表，把他们分别放在nextTable的i和i+n的位置上

• 如果这个位置是TreeBin节点（fh<0），也做一个反序处理，并且判断是否需要untreefi，把处理的结果分别放在nextTable的i和i+n的位置上

• 遍历过所有的节点以后就完成了复制工作，这时让nextTable作为新的table，并且更新sizeCtl为新容量的0.75倍 ，完成扩容。

再看一下多线程是如何完成的：

• 如果遍历到的节点是forward节点，就向后继续遍历，再加上给节点上锁的机制，就完成了多线程的控制。

• 多线程遍历节点，处理了一个节点，就把对应点的值set为forward，另一个线程看到forward，就向后遍历。

• 这样交叉就完成了复制工作。而且还很好的解决了线程安全的问题。 这个方法的设计实在是让我膜拜。

/**

• 一个过渡的table表，只有在扩容的时候才会使用

*/

private transient volatile Node<K,V>[] nextTable;

/**

• Moves and/or copies the nodes in each bin to new table. See

• above for explanation.

*/

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {

int n = tab.length, stride;

if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)

stride = MIN_TRANSFER_STRIDE; // subdivide range

if (nextTab == null) { // initiating

try {

@SuppressWarnings(“unchecked”)

Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];//构造一个nextTable对象 它的容量是原来的两倍

nextTab = nt;

} catch (Throwable ex) { // try to cope with OOME

sizeCtl = Integer.MAX_VALUE;

return;

}

nextTable = nextTab;

transferIndex = n;

}

int nextn = nextTab.length;

ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);//构造一个连节点指针 用于标志位

boolean advance = true;//并发扩容的关键属性 如果等于true 说明这个节点已经处理过

boolean finishing = false; // to ensure sweep before committing nextTab

for (int i = 0, bound = 0;