《Java高并发程序设计》学习 --4.4 无锁

对于并发控制，锁是一种悲观的策略。它总是假设每一次的临界区操作会产生冲突。如果有多个线程同时需要访问临界区资源，就宁可牺牲性能让线程进行等待，所以说锁会阻塞线程执行。而无锁是一种乐观的策略，它会假设对资源的访问是没有冲突的。无锁的策略使用一种叫做比较交换的技术（CAS Compare And Swap）来鉴别线程冲突，一旦检测到冲突产生，就重试当前操作直到没有冲突为止。

1）比较交换（CAS）

与锁相比，使用比较交换（CAS）会使程序看起来更加复杂一些。但由于其非阻塞性，它对死锁问题天生免疫，并且，线程间的相互影响也远远比基于锁的方式要小。更为重要的是，使用无锁的方式完全没有锁竞争带来的系统开销，也没有线程间频繁调度带来的开销，因此，它要比基于锁的方式拥有更优越的性能。

CAS算法的过程是这样的：它包含三个参数CAS(V,E,N)。V表示要更新的变量，E表示预期值，N表示新值。仅当V值等于E时，才会将V的值设为N，如果V值和E值不同，则说明已经有其他线程做了更新，则当前线程什么都不做。最后，CAS返回当前V的真实值。CAS操作是抱着乐观的态度进行的，它总认为自己可以完成操作。当多个线程同时使用CAS操作一个变量时，只有一个会胜出，并成功更新，其余均会失败。失败的线程不会被挂起，仅是被告知失败，并且允许再次尝试，当然也允许失败的线程放弃操作。基于这样的原理，CAS操作即使没有锁，也可以发现其他线程对当前线程的干扰，并进行恰当的处理。

2）无锁的线程安全整数：AtomicInteger

为了让Java程序员能够受益于CAS等CPU指令，JDK并发包中有一个atomic包，里面实现了一些直接使用CAS操作的线程安全的类型。

其中，最常用的一个类，应该就是AtomicInteger。可以把它看做是一个整数。但是与Integer不同，它是可变的，并且是线程安全的。对其进行任何修改等操作，都是用CAS指令进行的。这里简单列举一下AtomicInteger的一些主要方法：

//获取当前的值 public final int get() //设置当前值 public final void set(int newValue) //取当前的值，并设置新的值 public final int getAndSet(int newValue) //如果当前值为expect,则设置为update public final boolean compareAndSet(int expect,int update) //获取当前的值，并自增 public final int getAndIncrement() //获取当前的值，并自减 public final int getAndDecrement() //获取当前的值，并加上预期的值 public final int getAndAdd(int delta) //当前值增加delta,返回新值 public final int addAndGet(int delta) //当前值+1,返回新值 public final int incrementAndGet() //当前值-1,返回新值 public final int decrementAndGet()

AtomicInteger使用示例如下：

public class AtomicIntegerDemo {static AtomicInteger i = new AtomicInteger();public static class AddThred implements Runnable {@Overridepublic void run() {for(int k=0; k<10000; k++) {i.incrementAndGet();}}}public static void main(String[] args) throws InterruptedException {Thread[] ts = new Thread[10];for (int k = 0; k < 10; k++) {ts[k] = new Thread(new AddThred());}for(int k=0; k<10; k++) {ts[k].start();}for(int k=0; k<10; k++) {ts[k].join();}System.out.println(i);}
}

AtomicInteger.incrementAndGet()方法会使用CAS操作将自己加1，同时也会返回当前值。如果执行这段代码，程序输出100000。说明程序正常执行，没有错误。如果不是线程安全的，i的值应该会小于100000。

使用AtomicInteger会比使用锁具有更好的性能，测试代码如下：

public class UnLockExample1 {public static final int THREAD_NUM = 20;public static final int NUM = 1000000;//无锁变量区static AtomicInteger in = new AtomicInteger();//有锁变量区static int a = 0;public static class AddThread implements Runnable{@Overridepublic void run() {for (int k=0;k<NUM;k++) {in.incrementAndGet();  //详情参考  -->Java中的指针,Unsafe类}}}public static class lockExample implements Runnable{static final lockExample ue = new lockExample();@Overridepublic void run() {for (int i = 0; i < NUM; i++) {//加在这里是为了保持和CAS范围一致.synchronized (ue) {a += 1;}}}}public static void main(String[] a) throws InterruptedException {Thread[] threads = new Thread[THREAD_NUM];for (int i=0;i<THREAD_NUM ;i++){threads[i] = new Thread(new AddThread());}long s = System.currentTimeMillis();for (int i=0;i<THREAD_NUM ;i++)threads[i].start();for (int i=0;i<THREAD_NUM ;i++)threads[i].join();long e = System.currentTimeMillis();System.out.println("AddThread result : "+in.get()+" 耗时 : "+(e-s)+" ms");for (int i=0;i<THREAD_NUM ;i++){threads[i] = new Thread(new lockExample());}s = System.currentTimeMillis();for (int i=0;i<THREAD_NUM ;i++)threads[i].start();for (int i=0;i<THREAD_NUM ;i++)threads[i].join();e = System.currentTimeMillis();System.out.println("lockExample result : "+in.get()+" 耗时 : "+(e-s)+" ms");}
}

AddThread result : 20000000 耗时 : 1368 ms

lockExample result : 20000000 耗时 : 248 ms

为什么同步累加会比原子累加要快？

原子累加器的L1缓存失效比同步累加器高一个数量级。原子操作会导致缓存一致性问题，从而导致频繁的缓存行失效。但是这时同步累加器在一个CPU周期内反复的获取锁操作，缓存并没有失效。为什么我们会一直认为原子操作比加锁要快呢？文中的例子是很特别很特别的，正常业务场景下，我们累加过后，要经过很多业务代码逻辑才会再次去累加，这里已经跨过很多个CPU时间片了。从而同步累加器很难一直获取到锁，这中情况下，同步累加器即会有等待加锁的性能损失还会有缓存一致性带来的性能损失。所以在一般的情况下，同步累加器会慢很多。

incrementAndGet()的内部实现为：

public final int incrementAndGet() {for (;;) {int current = get();int next = current + 1;if (compareAndSet(current, next))return next;}
}

第2行的死循环是因为CAS操作未必是成功的，因此对于不成功的情况，我们就需要进行不断的尝试。第3行的get()取得当前的值，接着加1后得到新值next。这样，得到了CAS必需的两个参数：期望值以及新值。使用compareAndSet()方法将新值next写入，成功的条件是在写入的时刻，当前的值应该要等于刚刚取得的current。如果不是这样，说明AtomicInteger的值在第3行到第5行代码之间，又被其他线程修改了。当前线程看到的状态就是一个过期状态。因此，compareAndSet返回失败，需要下一次重试，直到成功。

和AtomicInteger类似的类还有AtomicLong用来代表long型，AtomicBoolean表示boolean型，AtomicReference表示对象引用。

3）Java中的指针：Unsafe类

public final boolean compareAndSet(int expect, int update) {return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}

这里的Unsafe封装了一些类似指针的从操作。compareAndSwapInt()方法是一个native方法，他的几个参数含义如下：

public final native boolean compareAndSwapInt(Object o, long offset, int expected, int x);

第一个参数o为给定的对象，offset为对象内的偏移量（其实就是一个字段到对象头部的偏移量，通过这个偏移量可以快速定位字段），excepted表示期望值，x表示要设置的值。如果指定的字段的值等于expected，那么就把它设置为x。

compareAndSwapInt()方法的内部，必须是使用CAS原子指令完成的。此外，Unsafe类还提供一些方法，主要有一下几个：

//获得给定对象偏移量上的int值
public native int getInt(Object o, long offset);
//设置给定对象偏移量上的int值
public native void putInt(Object o, long offset, int x);
//获得字段在对象中的偏移量
public native long objectFieldOffset(Field f);
//设置给定对象的int值，使用volatile语义
public native void putIntVolatile(Object o, long offset, int x);
//获得给定对象的int值，使用volatile语义
public native int getIntVolatile(Object o, long offset);
//和putIntVolatile()一样，但是它要求被操作字段就是volatile类型的

4）无锁的对象引用：AtomicReference

AtomicReference是对普通的对象引用，也就是它可以保证在修改对象引用时的线程安全性。

线程判断被修改对象是否可以正确写入的条件是对象的当前值和期望值是否一致。但有可能出现例外，当获得对象当前数据后，对象的值又恢复为旧值。这样，当前线程就无法正确判断这个对象究竟是否被修改过。

打一个比方，如果有一家蛋糕店，为了挽留客户，决定为贵宾卡里余额小于20元的客户一次性赠送20元。但条件是，每一位客户只能被赠送一次。

定义用户账户余额：

static AtomicReference<Integer> money = new AtomicReference<Integer>();
money.set(19);

接着，需要若干个后台线程，它们不断扫描数据，并为满足条件的客户充值。

for(int i=0; i<3; i++) {new Thread() {public void run() {while(true) {while(true) {Integer m = money.get();if(m<20) {if(money.compareAndSet(m, m+20)) {System.out.println("余额小于20元，充值成功，余额："+ money.get() + "元");break;} else {break;}}}}};}.start();

此时，如果很不幸，用户正好正在进行消费，就在赠予金额到账的同时，他进行了一次消费，似的金额又小于20元，并且正好累计消费了20元。似的消费、赠予后的金额等于消费前、赠予前的金额。这时，后台的赠予进行就会误以为这个账户还没有赠予，所以，存在多次被赠予的可能。下面模拟了这个消费线程：

new Thread() {public void run() {for (int j = 0; j < 100; j++) {while(true) {Integer m = money.get();if(m>10) {System.out.println("大于10元");if(money.compareAndSet(m, m-10)) {System.out.println("成功消费10元，余额：" + money.get());break;}} else {System.out.println("没有足够金额");break;}}try {Thread.sleep(100);} catch (InterruptedException e) {}}};
}.start();

上述代码，消费者只要贵宾卡里的钱大于10元，就会立即进行一次10元的消费。执行上述代码，得到的输出如下：

余额小于20元，充值成功，余额：39元
大于10元
成功消费10元，余额：29
大于10元
成功消费10元，余额：19
余额小于20元，充值成功，余额：39元
大于10元
成功消费10元，余额：29
大于10元
成功消费10元，余额：39
余额小于20元，充值成功，余额：39元

从这一段输出中，可以看到，这个账户被先后反复多次充值。其原因正是因为账户余额被反复修改，修改后的值等于原有的值，使得CAS操作无法正确判断当前数据状态。

5）带有时间戳的对象引用：AtomicStampedReference

AtomicStampedReference解决了上述问题，其内部不仅维护了对象值，还维护了一个时间戳。当AtomicStampedReference对应的数值被修改时，除了更新数据本身外，还必须要更新时间戳。当AtomicStampedReference设置对象值时，对象值以及时间戳都必须满足期望值，写入才会成功。因此，即使对象值被反复读写，写回原值，只要时间戳发送变化，就能防止不恰当的写入。

AtomicStampedReference的几个API在AtomicReference的基础上新增了有关时间戳的信息：

//比较设置参数依次为：期望值  写入新值  期望时间戳  新时间戳
public boolean compareAndSet(V   expectedReference,V   newReference,int expectedStamp,int newStamp)
//获得当前对象引用
public V getReference()
//获得当前时间戳
public int getStamp()
//设置当前对象引用和时间戳
public void set(V newReference, int newStamp) 使用AtomicStampedReference来修正那个贵宾卡充值的问题：
public class AtomicStampedReferenceDemo {
static AtomicStampedReference<Integer> money = new AtomicStampedReference<Integer>(19,0);public static void main(String[] args) {for(int i=0; i<3; i++) {final int timestamp = money.getStamp();new Thread() {public void run() {while(true) {while(true) {Integer m = money.getReference();if(m<20) {if(money.compareAndSet(m, m+20,timestamp,timestamp+1)) {System.out.println("余额小于20元，充值成功，余额："+ money.getReference() + "元");break;} else {break;}}}}};}.start();}new Thread() {public void run() {for (int j = 0; j < 100; j++) {while(true) {int timestamp = money.getStamp();Integer m = money.getReference();if(m>10) {System.out.println("大于10元");if(money.compareAndSet(m, m-10,timestamp,timestamp+1)) {System.out.println("成功消费10元，余额：" + money.getReference());break;}} else {System.out.println("没有足够金额");break;}}try {Thread.sleep(100);} catch (InterruptedException e) {// TODO: handle exception}}};}.start();}
}

执行上述代码，可以得到以下输出：

余额小于20元，充值成功，余额：39元
大于10元
成功消费10元，余额：29
大于10元
成功消费10元，余额：19
大于10元
成功消费10元，余额：9
没有足够金额

可以看到，账户只被赠予了一次。

6）数组也能无锁：AtomicIntegerArray

当前可用的原子数组有：AtomicIntegerArray、AtomicLongArray、AtomicReferenceArray，分别表示整数数组、long型数组和普通对象数组。

以AtomicIntegerArray为例，展示原子数组的使用方式。

AtomicIntegerArray本质上是对int[]类型的封装，使用Unsafe类通过CAS的方式控制int[]在多线程的安全性。它提供以下几个核心API：

//获得数组第i个下标的元素
public final int get(int i)
//获得数组的长度
public final int length()
//将数组第i个下标设置为newValue，并返回旧的值
public final int getAndSet(int i, int newValue)//进行CAS操作，如果第i个下标的元素等于expect，则设置为update，设置成功返回truepublic final boolean compareAndSet(int i, int expect, int update)//将第i个下标元素加 1public final int getAndIncrement(int i)//将第i个下标的元素减1public final int getAndDecrement(int i)//将第i个下标的元素增加delta(delta可以是负数)public final int getAndAdd(int i, int delta)

下面一个简单示例，展示AtomicIntegerArray的使用：

public class AtomicIntegerArrayDemo {	static AtomicIntegerArray arr = new AtomicIntegerArray(10);public static class AddThread implements Runnable {@Overridepublic void run() {for(int k=0; k<10000; k++) {arr.getAndIncrement(k%arr.length());}}}public static void main(String[] args) throws InterruptedException {Thread[] ts = new Thread[10];for(int k=0; k<10; k++) {ts[k] = new Thread(new AddThread());}for(int k=0; k<10; k++) {ts[k].start();}for(int k=0; k<10; k++) {ts[k].join();}System.out.println(arr);}
}

如果线程安全，数组内10个元素的值必然都是10000。反之，如果线程不安全，则部分或者全部数值会小于10000。

7）让普通变量也享受原子操作：AtomicIntegerFieldUpdater

在原子包里有一个使用的工具类AtomicIntegerFieldUpdater。它可以让你在不改动（或者极少改动）原有代码的基础上，让普通的变量也享受CAS操作带来的线程安全性。

根据数据类型不同，这个Updater有三种，分别是AtomicIntegerFieldUpdater、AtomicLongFieldUpdater和AtomicReferenceFieldUpdater，它们分别对int、long和普通对象进行CAS修改。

现在思考一个场景。假设某地要进行一次选举。现在模拟这个投票场景，如果选民投了候选人一票，就记为1，否则记为0。最终的选票显然就是所有数据的简单求和。

public class AtomicIntegerFieldUpdaterDemo {public static class Candidate {int id;volatile int score;}public final static AtomicIntegerFieldUpdater<Candidate> scoreUpdater = AtomicIntegerFieldUpdater.newUpdater(Candidate.class, "score");public static AtomicInteger allScore = new AtomicInteger(0);public static void main(String[] args) throws InterruptedException {final Candidate stu = new Candidate();Thread[] t = new Thread[10000];for(int i=0; i<10000; i++) {t[i] = new Thread() {@Overridepublic void run() {if(Math.random() > 0.4) {scoreUpdater.incrementAndGet(stu);allScore.incrementAndGet();}}};t[i].start();}for(int i=0; i<10000; i++) {t[i].join();}System.out.println("score=" + stu.score);System.out.println("allScore=" + allScore);}
}

运行这段程序，最终的Candidate.score总是和allScore绝对相等。这说明AtomicIntegerFieldUpdater很好地保证了Candidate.score的线程安全。

虽然AtomicIntegerFieldUpdater很好用，但有以下几个注意事项：

第一， Updater只能修改它可见范围的变量。因为Updater是通过反射得到的这个变量。如果变量不可见，会出错。比如score设置为private,就不行。

第二，为了确保变量被正确的读取，必须是volatile修饰。

第三，由于CAS操作会通过对象实例中的偏移量直接进行赋值，因此，它不支持static字段(public native long objectFieldOffset()不支持静态变量) 。（ 静态变量存储在静态存储区，程序启动时就分配空间，程序退出时释放。普通成员变量在类实例化时分配空间，释放类的时候释放空间，存储在栈或堆中。）

8）无锁的Vector实现

我们将这个无锁的Vector称为LockFreeVector。它的特点是可以根据需求动态扩展其内部空间。在这里，我们使用二维数组来表示LockFreeVector的内部存储，如下：

private final AtomicReferenceArray<AtomicReferenceArray<E>> buckets; 

变量buckets存放所有的内部元素。从定义上看，它是一个保存着数组的数组，也就是通常的二维数组。特别之处在于这些数组都是使用CAS的原子数组。为什么使用二维数组去实现一个一维的Vector呢？这是为了将来Vector进行动态扩展时可以更加方便。我们知道，AtomicReferenceArray内部使用Object[]来进行实际数据的存储，这使得动态空间增加特别的麻烦，因此使用二维数组的好处就是为将来增加新的元素。

此外，为了更有序的读写数组，定义一个称为Descriptor的元素。它的作用是使用CAS操作写入新数据。

static class Descriptor<E> {  public int size;  volatile WriteDescriptor<E> writeop;  public Descriptor(int size, WriteDescriptor<E> writeop) {  this.size = size;  this.writeop = writeop;  }  public void completeWrite() {  WriteDescriptor<E> tmpOp = writeop;  if (tmpOp != null) {  tmpOp.doIt();  writeop = null; // this is safe since all write to writeop use  // null as r_value.  }  }  
}  static class WriteDescriptor<E> {  public E oldV;  public E newV;  public AtomicReferenceArray<E> addr;  public int addr_ind;  public WriteDescriptor(AtomicReferenceArray<E> addr, int addr_ind,  E oldV, E newV) {  this.addr = addr;  this.addr_ind = addr_ind;  this.oldV = oldV;  this.newV = newV;  }  public void doIt() {  addr.compareAndSet(addr_ind, oldV, newV);  }  
} 

上述代码第4行定义的Descriptor构造函数接收2个参数，第一个为整个Vector的长度，第2个为一个writer。最终，写入数据是通过writer进行的（通过completeWrite()方法）。第24行，WriteDescriptor的构造函数接收4个参数。第一个参数addr表示要修改的原子数组，第二个参数为要写入的数组索引位置，第三个oldV为期望值，第4个newV为需要写入的值。

在构造LockFreeVector时，显然需要将buckets和descriptor进行初始化。  

public LockFreeVector() {  buckets = new AtomicReferenceArray<AtomicReferenceArray<E>>(N_BUCKET);  buckets.set(0, new AtomicReferenceArray<E>(FIRST_BUCKET_SIZE));  descriptor = new AtomicReference<Descriptor<E>>(new Descriptor<E>(0,  null));  
}  

在这里N_BUCKET为30,也就是说这个buckets里面可以存放一共30个数组（由于数组无法动态增长，因此数组总数也就不能超过30个）。并且将第一个数组的大小为FIRST_BUCKET_SIZE为8。到这里，大家可能会有一个疑问，如果每个数组8个元素，一共30个数组，那岂不是一共只能存放240个元素吗？

如果大家了解JDK内的Vector实现，应该知道，Vector在进行空间增长时，默认情况下，每次都会将总容量翻倍。因此，这里也借鉴类似的思想，每次空间扩张，新的数组的大小为原来的2倍（即每次空间扩展都启用一个新的数组），因此，第一个数组为8，第2个就是16，第3个就是32。以此类推，因此30个数组可以支持的总元素达到。

这数值已经超过了2^33，即在80亿以上。因此，可以满足一般的应用。

当有元素需要加入LockFreeVector时，使用一个名为push_back()的方法，将元素压入Vector最后一个位置。这个操作显然就是LockFreeVector的最为核心的方法，也是最能体现CAS使用特点的方法，它的实现如下：

public void push_back(E e) {  Descriptor<E> desc;  Descriptor<E> newd;  do {  desc = descriptor.get();  desc.completeWrite();  int pos = desc.size + FIRST_BUCKET_SIZE;  int zeroNumPos = Integer.numberOfLeadingZeros(pos);  int bucketInd = zeroNumFirst - zeroNumPos;  if (buckets.get(bucketInd) == null) {  int newLen = 2 * buckets.get(bucketInd - 1).length();  if (debug)  System.out.println("New Length is:" + newLen);  buckets.compareAndSet(bucketInd, null,  new AtomicReferenceArray<E>(newLen));  }  int idx = (0x80000000>>>zeroNumPos) ^ pos;  newd = new Descriptor<E>(desc.size + 1, new WriteDescriptor<E>(  buckets.get(bucketInd), idx, null, e));  } while (!descriptor.compareAndSet(desc, newd));  descriptor.get().completeWrite();  
}  

可以看到，这个方法主体部分是一个do-while循环，用来不断尝试对descriptor的设置。也就是通过CAS保证了descriptor的一致性和安全性。在第23行，使用descriptor将数据真正地写入数组中。这个descriptor写入的数据由20~21行构造的WriteDescriptor决定。

在循环最开始（第5行），使用descriptor先将数据写入数组，是为了防止上一个线程设置完descriptor后（22行），还没来得及执行第23行的写入，因此，做一次预防性的操作。

因为限制要将元素e压入Vector，因此，我们必须首先指定这个e应该放在哪个位置。由于目前使用了二维数组，因此我们自然需要知道e所在哪个数组（buckets中的下标位置）和数组中的下标。

第8~10行通过当前Vector的大小（desc.size），计算新的元素应该落入哪个数组。这里使用了位运算进行计算。

LockFreeVector每次都会扩容。它的第一个数组长度为8，第2个就是16，第3个就是32，依次类推。它们的二进制表示如下：

      00000000 00000000 00000000 00001000：第一个数组大小，28个前导零。00000000 00000000 00000000 00010000：第二个数组大小，27个前导零。00000000 00000000 00000000 00100000：第三个数组大小，26个前导零。00000000 00000000 00000000 01000000：第四个数组大小，25个前导零。

它们之和就是整个LockFreeVector的总大小，因此，如果每一个数组都恰好填满，那么总大小应该类似如下的值（以4个数组为例）。

      00000000 00000000 00000000 01111000：4个数组都恰好填满时的大小。

导致这个数字进位的最小条件，就是加上二进制的1000。而这个数字整好是8（FIRST_BUCKET_SIZE就是8）。这就是第8行代码的意义。它可以使得数组大小发生一次二进制进位（如果不进位说明还在第一个数组中），进位后前导零的数量就会发生变化。而元素所在的数组，和pos（第8行定义的比变量）的前导零直接相关。每进行一次数组扩容，它的前导零就会减1。如果从来没有扩容过，它的前导零就是28个。以后，逐级减1。这就是第9行获得pos前导零的原因。第10行，通过pos的前导零可以立即定位使用哪个数组（也就是得到了bucketInd的值）。

第11行，判断这个数组是否存在。如果不存在，则创建这个数组，大小为前一个数组的两倍，并把它设置到buckets中。

接着再看一下元素没有恰好填满的情况。

      00000000 00000000 00000000 00001000：第一个数组大小，28个前导零00000000 00000000 00000000 00010000：第二个数组大小，27个前导零。00000000 00000000 00000000 00100000：第三个数组大小，26个前导零。00000000 00000000 00000000 00000001：第四个数组大小，只有一个前导零。

那么总大小如下：

      00000000 00000000 00000000 00111001：元素总个数

总个数加上二进制1000后，得到：

      00000000 00000000 00000000 01000001

显然，通过前导零可以定位到第4个数组。而剩余位，显然就表示元素在当前数组内偏移量（也就是数组下标）。根据这个理论，就可以通过pos计算这个元素应该放在给定数组的哪个位置。通过第19行代码，获得pos的除了第一位数字1以外的其他位的数值。因此，pos的前导零可以表示元素所在的数组，而pos的后面几位，则表示元素所在这个数组中的位置。由此，第19行代码就取得了元素所在位置idx。

到此，我们就已经得到新元素位置的全部信息，剩下的就是将这些信息传递给Descriptior让它在给定的位置把元素e安置上去即可。这里，通过CAS操作，保证写入正确性。

下面来看一下get()操作的实现：

@Override
public E get(int index) {int pos = index + FIRST_BUCKET_SIZE;int zeroNumPos = Integer.numberOfLeadingZeros(pos);int bucketInd = ZERO_NUM_FIRST - zeroNumPos;int idx = (MARK_FIRST_BIT >>> zeroNumPos) ^ pos;return buckets.get(bucketInd).get(idx);
}

这get()的实现中，第3~6行使用了相同的算法获得所需元素的数组以及数组中的索引下标。这里简单地通过buckets定位到对应的元素即可。

9）让线程之间互相帮助：细看SynchronousQueue的实现

在对线程池的介绍中，提到了一个非常特殊的等待队列SynchronousQueue。SynchronousQueue的容量为0，任何一个对SynchronousQueue的写需要等待一个对SynchronousQueue的读，反之亦然。因此，SynchronousQueue与其说是一个队列，不如说是一个数据交换通道。那SynchronousQueue的其妙功能是如何实现的呢？

既然我打算在这一节中介绍它，那么SynchronousQueue比如和无锁的操作脱离不了关系。实际上SynchronousQueue内部也正是大量使用了无锁工具。

对SynchronousQueue来说，它将put()和take()两个功能截然不同的操作抽象为一个共通的方法Transferer.transfer()。从字面上看，这就是数据传递的意思。它的完整签名如下：

Object transfer(Object e, boolean timed, long nanos) 

当参数e为非空时，表示当前操作传递给一个消费者，如果为空，则表示当前操作需要请求一个数据。timed参数决定是否存在timeout时间，nanos决定了timeout的时长。如果返回值非空，则表示数据以及接受或者正常提供，如果为空，则表示失败（超时或者中断）。

SynchronousQueue内部会维护一个线程等待队列。等待队列中会保存等待线程以及相关数据的信息。比如，生产者将数据放入SynchronousQueue时，如果没有消费者接受，那么数据本身和线程对象都会打包在队列中等待（因为SynchronousQueue容积为0，没有数据可以正常放入）。

Transferer.transfer()函数的实现是SynchronousQueue的核心，它大体上分为三个步骤：

1、如果等待队列为空，或者队列中节点的类型和本次操作是一致的，那么将当前操作压入队列等待。比如，等待队列中是读线程等待，本次操作也是读，因此这2个读都需要等待。进入等待队列的线程可能会被挂起，它们会等待一个“匹配”操作。

2、如果等待队列中的元素和本次操作是互补的（比如等待操作是读，而本次操作是写），那么就插入一个“完成”状态的节点，并且让他“匹配”到一个等待节点上。接着弹出这2个节点，并且使得对应的2个线程继续执行。

3、如果线程发现等待队列的节点就是“完成”节点。那么帮助这个节点完成任务。其流程和步骤2是一致的。

步骤1的实现如下（代码参考JDK 7u60）：

SNode h = head;  
if (h == null || h.mode == mode) {                  // 如果队列为空，或者模式相同  if (timed && nanos <= 0) {                   // 不进行等待  if (h != null && h.isCancelled())  casHead(h, h.next);                 // 处理取消行为  else  return null;  } else if (casHead(h, s = snode(s, e, h, mode))) {  SNode m = awaitFulfill(s, timed, nanos);    //等待，直到有匹配操作出现  if (m == s) {                               // 等待被取消  clean(s);  return null;  }  if ((h = head) != null && h.next == s)  casHead(h, s.next);                 // 帮助s的 fulfiller  return (mode == REQUEST) ? m.item : s.item;  }  
}  

上述代码中，第1行SNode表示等待队列中的节点。内部封装了当前线程、next节点、匹配节点、数据内容等信息。第2行，判断当前等待队列为空，或者队列中元素的模式与本次操作相同（比如，都是读操作，那么都必须要等待）。第8行，生成一个新的节点并置于队列头部，这个节点就代表当前线程。如果入队成功，则执行第9行awaitFulfill()函数。该函数会进行自旋等待，并最终挂起当前线程。直到一个与之对应的操作产生，将其唤醒。线程被唤醒后（表示已经读取到数据或者自己产生的数据已经被别的线程读取），在14~15行尝试帮助对应的线程完成两个头部节点的出队操作（这仅仅是友情帮助）。并在最后，返回读取或者写入的数据（第16行）。

步骤2的实现如下：

} else if (!isFulfilling(h.mode)) {             //是否处于fulfill状态  if (h.isCancelled())                // 如果以前取消了  casHead(h, h.next);             // 弹出并重试  else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {  for (;;) {                        // 一直循环直到匹配（match）或者没有等待者了  SNode m = s.next;           // m 是 s的匹配者（match）  if (m == null) {                // 已经没有等待者了  casHead(s, null);           // 弹出fulfill节点  s = null;                   // 下一次使用新的节点  break;                  // 重新开始主循环  }  SNode mn = m.next;  if (m.tryMatch(s)) {  casHead(s, mn);         // 弹出s 和 m  return (mode == REQUEST) ? m.item : s.item;  } else                      // match 失败  s.casNext(m, mn);       // 帮助删除节点  }  }  
}  

上述代码中，首先判断头部节点是否处于fulfill模式。如果是，则需要进入步骤3。否则，将试自己为对应的fulfill线程。第4行，生成一个SNode元素，设置为fulfill模式并将其压入队列头部。接着，设置m（原始的队列头部）为s的匹配节点（第13行），这个tryMatch()操作将会激活一个等待线程，并将m传递给那个线程。如果设置成功，则表示数据投递完成，将s和m两个节点弹出即可（第14行）。如果tryMatch()失败，则表示已经有其他线程帮我完成了操作，那么简单得删除m节点即可（第17行），因为这个节点的数据已经被投递，不需要再次处理，然后，再次跳转到第5行的循环体，进行下一个等待线程的匹配和数据投递，直到队列中没有等待线程为止。

步骤3：如果线程在执行时，发现头部元素恰好是fulfill模式，它就会帮助这个fulfill节点尽快被执行：

} else {                                       // 帮助一个 fulfiller  SNode m =h.next;                           // m 是 h的 match  if (m ==null)                              // 没有等待者  casHead(h,null);                       // 弹出fulfill节点  else {  SNode mn =m.next;  if(m.tryMatch(h))                      // 尝试 match  casHead(h, mn);                     // 弹出 h 和 m  else                                   // match失败  h.casNext(m,mn);                   // 帮助删除节点  }  
}  

上述代码的执行原理和步骤2是完全一致的。唯一的不同是步骤3不会返回，因为步骤3所进行的工作是帮助其他线程尽快投递它们的数据。而自己并没有完成对应的操作，因此，线程进入步骤3后，再次进入大循环体（代码中没有给出），从步骤1开始重新判断条件和投递数据。

从整个数据投递的过程中可以看到，在SynchronousQueue中，参与工作的所有线程不仅仅是竞争资源的关系。更重要的是，它们彼此之间还会相互帮助。在一个线程内部，可能会帮助其他线程完成它们的工作。这种模式可以更大程度上减少饥饿的可能，提高系统整体的并行度。

5）有关死锁的问题

死锁就是两个或者多个线程，相互占用对方需要的资源，而都不进行释放，导致彼此之间都相互等待对方释放资源，产生了无限制等待的现象。死锁一旦发生，如果没有外力接入，这种等待将永远存在，从而对线程产生严重的影响。

下面用一个简单的例子模拟哲学家就餐问题的过程：

public class DeadLock extends Thread {protected Object tool;static Object fork1 = new Object();static Object fork2 = new Object();public DeadLock(Object obj) {this.tool = obj;if(tool == fork1) {this.setName("A");}if(tool == fork2) {this.setName("B");}}@Overridepublic void run() {if(tool == fork1) {synchronized (fork1) {try {Thread.sleep(500);} catch (Exception e) {e.printStackTrace();}synchronized (fork2) {System.out.println("哲学家A开始吃饭了");}}}if(tool == fork2) {synchronized (fork2) {try {Thread.sleep(500);} catch (Exception e) {e.printStackTrace();}synchronized (fork1) {System.out.println("哲学家B开始吃饭了");}}}}public static void main(String[] args) throws InterruptedException {DeadLock A = new DeadLock(fork1);DeadLock B = new DeadLock(fork2);A.start();B.start();Thread.sleep(1000);}
}

上述代码模拟了两个哲学家互相等待对方的叉子。哲学家A先占用叉子1，哲学家B占用叉子2，接着他们就互相等待，都没有办法获得两个叉子用餐。

如果在实际环境中，遇到了这种情况，通常的表现就是相关的进程不再工作，并且CPU占用率为0（因为死锁的线程不占用CPU）。想确认问题，需要使用JDK提供的工具。

首先，可以使用jps命令得到java进程的进程ID，接着使用jstack命令得到线程的线程堆栈：

Microsoft Windows [版本 6.1.7601]

版权所有 (c) 2009 Microsoft Corporation。保留所有权利。

C:\Users\haoning>jps
9008 DeadLock
8788 JpsC:\Users\haoning>jstack 9008
2017-01-06 20:28:10
Full thread dump Java HotSpot(TM) 64-Bit Server VM (24.0-b56 mixed mode):"DestroyJavaVM" prio=6 tid=0x000000000049d800 nid=0x2fe4 waiting on condition [0
x0000000000000000]java.lang.Thread.State: RUNNABLE"B" prio=6 tid=0x000000000a4fa800 nid=0x1dc4 waiting for monitor entry [0x000000
000acdf000]java.lang.Thread.State: BLOCKED (on object monitor)at cn.guet.parallel.DeadLock.run(DeadLock.java:42)- waiting to lock <0x00000000eae97dc8> (a java.lang.Object)- locked <0x00000000eae97dd8> (a java.lang.Object)"A" prio=6 tid=0x000000000a4f5800 nid=0x2acc waiting for monitor entry [0x000000
000abbf000]java.lang.Thread.State: BLOCKED (on object monitor)at cn.guet.parallel.DeadLock.run(DeadLock.java:29)- waiting to lock <0x00000000eae97dd8> (a java.lang.Object)- locked <0x00000000eae97dc8> (a java.lang.Object)
Found one Java-level deadlock:
=============================
"B":waiting to lock monitor 0x000000000a4fc308 (object 0x00000000eae97dc8, a java.
lang.Object),which is held by "A"
"A":waiting to lock monitor 0x00000000088c3918 (object 0x00000000eae97dd8, a java.
lang.Object),which is held by "B"Java stack information for the threads listed above:
===================================================
"B":at cn.guet.parallel.DeadLock.run(DeadLock.java:42)- waiting to lock <0x00000000eae97dc8> (a java.lang.Object)- locked <0x00000000eae97dd8> (a java.lang.Object)
"A":at cn.guet.parallel.DeadLock.run(DeadLock.java:29)- waiting to lock <0x00000000eae97dd8> (a java.lang.Object)- locked <0x00000000eae97dc8> (a java.lang.Object)Found 1 deadlock.

上面显示了jstack部分输出。可以看到，哲学家A和哲学家B两个线程发生了死锁。并且在最后，可以看到两者相互等待的锁的ID。同时，死锁的两个线程均处于BLOCK状态。

注：本篇博客内容摘自《 Java 高并发程序设计》