【实战Java高并发程序设计6】挑战无锁算法-无锁的Vector实现

转载请注明出处,感谢大家的支持!
本文来自优优码:http://www.uucode.net/201602/bingfa6

【实战Java高并发程序设计】连载1–Java中的指针:Unsafe类

【实战Java高并发程序设计】连载2–无锁的对象引用:AtomicReference

【实战Java高并发程序设计】连载3:带有时间戳的对象引用:AtomicStampedReference

【实战Java高并发程序设计】连载4:数组也能无锁:AtomicIntegerArray

【实战Java高并发程序设计】连载5:让普通变量也享受原子操作

我们已经比较完整得介绍了有关无锁的概念和使用方法。相对于有锁的方法,使用无锁的方式编程更加考验一个程序员的耐心和智力。但是,无锁带来的好处也是显而易见的,第一,在高并发的情况下,它比有锁的程序拥有更好的性能;第二,它天生就是死锁免疫的。就凭借这2个优势,就值得我们冒险尝试使用无锁的并发。

这里,我想向大家介绍一种使用无锁方式实现的Vector。通过这个案例,我们可以更加深刻地认识无锁的算法,同时也可以学习一下有关Vector实现的细节和算法技巧。(在本例中,讲述的无锁Vector来自于amino并发包)

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

private final AtomicReferenceArray<AtomicReferenceArray<E>> buckets;

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

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

01 static class Descriptor<E> {

02    public int size;

03    volatile WriteDescriptor<E> writeop;

04    public Descriptor(int size, WriteDescriptor<E> writeop) {

05             this.size = size;

06             this.writeop = writeop;

07    }

08    public void completeWrite() {

09             WriteDescriptor<E> tmpOp = writeop;

10             if (tmpOp != null) {

11                      tmpOp.doIt();

12                      writeop = null; // this is safe since all write to writeop use

13                      // null as r_value.

14             }

15    }

16 }

17

18 static class WriteDescriptor<E> {

19    public E oldV;

20    public E newV;

21    public AtomicReferenceArray<E> addr;

22    public int addr_ind;

23

24    public WriteDescriptor(AtomicReferenceArray<E> addr, int addr_ind,

25                      E oldV, E newV) {

26             this.addr = addr;

27             this.addr_ind = addr_ind;

28             this.oldV = oldV;

29             this.newV = newV;

30    }

31

32    public void doIt() {

33             addr.compareAndSet(addr_ind, oldV, newV);

34    }

35 }

上述代码第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使用特点的方法,它的实现如下:

01 public void push_back(E e) {

02    Descriptor<E> desc;

03    Descriptor<E> newd;

04    do {

05             desc = descriptor.get();

06             desc.completeWrite();

07

08             int pos = desc.size + FIRST_BUCKET_SIZE;

09             int zeroNumPos = Integer.numberOfLeadingZeros(pos);

10             int bucketInd = zeroNumFirst – zeroNumPos;

11             if (buckets.get(bucketInd) == null) {

12                      int newLen = 2 * buckets.get(bucketInd – 1).length();

13                      if (debug)

14                                System.out.println(“New Length is:” + newLen);

15                      buckets.compareAndSet(bucketInd, null,

16                                         new AtomicReferenceArray<E>(newLen));

17             }

18

19             int idx = (0×80000000>>>zeroNumPos) ^ pos;

20             newd = new Descriptor<E>(desc.size + 1, new WriteDescriptor<E>(

21                                buckets.get(bucketInd), idx, null, e));

22    } while (!descriptor.compareAndSet(desc, newd));

23    descriptor.get().completeWrite();

24 }

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

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

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

第8到10行通过当前Vector的大小(desc.size),计算新的元素应该落入哪个数组。这里使用了位运算进行计算。这几行代码看起来也许你会觉得有些奇怪,我的解释如下:

之前说过,LockFreeVector每次都会成倍的扩容。它的第1个数组长度为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行,判断这个数组是否存在。如果不存在,则创建这个数组,大小为前1个数组的2倍,并把它设置到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位数字1以外的其他位的数值。因此,pos的前导零可以表示元素所在的数组,而pos的后面几位,则表示元素所在这个数组中的位置。由此,第19行代码就取得了元素的所在位置idx。

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

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

1 @Override

2 public E get(int index) {

3      int pos = index + FIRST_BUCKET_SIZE;

4      int zeroNumPos = Integer.numberOfLeadingZeros(pos);

5      int bucketInd = zeroNumFirst – zeroNumPos;

6      int idx = (0×80000000>>>zeroNumPos) ^ pos;

7      return buckets.get(bucketInd).get(idx);

8 }

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

这样,对于Vector来说2个重要的方法就已经实现了。其他方法也是非常类似的,这里就不再详细讨论了。

参考本书:

javabing

2 个评论

  1. 您好,最近正在读《实战JAVA高并发程序设计》,发现了点问题
    (1)第17页的第一行,一辆汽车行驶在相距60公里的城市,书中是 相聚
    (2)第33页中,继承Thread类,应该是重写run方法,书中是重载run方法

发表评论

您的电子邮件地址将不会被公开. 必填昵称和邮箱 *

*

您可以使用这些HTML标签和属性: <a href="" title=""> <abbr title=""> <acronym title=""> <b> <blockquote cite=""> <cite> <code> <del datetime=""> <em> <i> <q cite=""> <strike> <strong>

你必须启用javascript
滚动到顶部
备案号:浙ICP备08112675号-4 浙公网安备 33010502001953号