关于java线程(4)----JUC之原子操作

碧海山城

浏览: 189867 次
性别:
来自: 杭州

最近访客更多访客>>

winting

ibnhuang

laosg

tanhaiyuan

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

java live

Java 理论与实践: 流行的原子

Java 理论与实践: 非阻塞算法简介

在java中确保共享变量线程安全的传统方式是使用同步，同步可以确定访问一组变量的所有线程都将拥有对这些变量的独占访问权（原子性），并且其他线程获得该锁定时，将可以看到对这些变量的更改（可见性）。但是锁的代价太昂贵，特别是在竞争很厉害的时候，影响吞吐量。

基本变量的原子访问+volatile

原子操作以为这不能中断的，要么全部成功，要么全部失败！Java中有些操作时原子的：

1.对原始类型的读写（除了long和double）

2..对所有volatile类型的读写

但是他们都不能保证一些简单复合操作的原子性！

A.1 CAS

CAS 操作包含三个操作数 —— 内存位置（V）、预期原值（A）和新值(B)

CAS 原理:

我认为位置 V 应该包含值 A；如果包含该值，则将 B 放到这个位置；否则，不要更改该位置，只告诉我这个位置现在的值即可。他是非阻塞的。

从某种意义上来看，简单的复合操作，不管是getAndInc和getAndDec还有IncAndGet、DecAndGet等等，其实都可以归结为一个CAS操作，比如getAndInc，在for循环内话哦去原值，并且+1，通过cas设置结果，如果成功的话返回，否则继续！

基于 CAS 的并发算法称为 无锁定算法，因为线程不必再等待锁定。无论 CAS 操作成功还是失败，在任何一种情况中，它都在可预知的时间内完成。如果 CAS 失败，调用者可以重试 CAS 操作或采取其他适合的操作。

A.2 CAS的ABA问题

CAS, 语义上, 有一个漏洞, 当第一次读取V的A值, 此时, 内存V的值变为B值, 然后在未执行CAS前, 又变回了A值.此时, CAS再执行时, 会判断其正确的, 并进行赋值.

这种判断值的方式来断定内存是否被修改过, 针对某些问题, 是不适用的.

jdk 1.5并发包提供了AtomicStampedReference(有标记的原子引用)类, 通过控制变量值的版本来保证CAS正确性.其实, 大部分通过值的变化来CAS, 已经够用了.

A3. CAS和锁

先来看个使用原子类的demo：

public class TestAtomicInteger {
	final AtomicInteger ai;
	public TestAtomicInteger(int total){
		ai=new AtomicInteger(total);
	}
	
	public static void main(String[] agrs) throws InterruptedException{
		final TestAtomicInteger ti=new TestAtomicInteger(100000);
		//搞10个线程去更新
		ExecutorService es=Executors.newFixedThreadPool(10);
		List<Callable<String>> tasks=new ArrayList<Callable<String>>();
		for(int i=0;i<10;i++){
			tasks.add(new Callable<String>() {
				public String call() throws Exception{
					ti.incerment();
					return "";
				}
			});
		}

		es.invokeAll(tasks);
		es.shutdown();
	}
	
	String incerment(){
		int aii=ai.get();
		Thread.yield();
		if(ai.compareAndSet(aii, aii+1)){
			System.out.println(Thread.currentThread().getName()+"execute successful! "+aii);
		}else{
			System.out.println(Thread.currentThread().getName()+" execute  failed!");
		}
	
		return "";	
	}
}

结果：

=================================

pool-1-thread-1execute successful! 100000

pool-1-thread-3execute successful! 100002

pool-1-thread-4 execute failed!

pool-1-thread-2execute successful! 100001

pool-1-thread-8execute successful! 100003

pool-1-thread-9 execute failed!

pool-1-thread-5 execute failed!

pool-1-thread-7execute successful! 100004

pool-1-thread-10 execute failed!

pool-1-thread-6 execute failed!

可以看到，有一些更新失败了！

为了确保正常更新，可能得将CAS操作放到for循环里，从语法结构上来看，使用CAS比使用锁更加复杂，得考虑失败的情况（锁会挂起线程，直到恢复）；但是基于CAS的原子操作，在性能上基本超过了基于锁的计数器，即使只有很小的竞争或者不存在竞争！

在轻度到中度的争用情况下，非阻塞算法的性能会超越阻塞算法，因为 CAS 的多数时间都在第一次尝试时就成功，而发生争用时的开销也不涉及线程挂起和上下文切换，只多了几个循环迭代。没有争用的 CAS 要比没有争用的锁便宜得多（这句话肯定是真的，因为没有争用的锁涉及 CAS 加上额外的处理，加锁至少需要一个CAS，在有竞争的情况下，需要操作队列，线程挂起，上下文切换），而争用的 CAS 比争用的锁获取涉及更短的延迟。

在java5之前，除非写本机代码通过jni，否则不能在java代码里实现CAS，Java5引入了底层的支持：

/** 
*比较并更新对象的某一个整数类型的域 
*@param obj 被操作的对象 
*@param fieldoffset 被操作的域在对象中的偏移量 
*@param expect 域的期望值 
*@param update 域的更新值 
*/  

unsafe.compareAndSwapInt(this, valueOffset, expect, update);

B原子类4组

原子变量类共有12个类，4组：

B.1 计数器：AtomicInteger/Long/Boolean/Reference

AtomicInteger和Long还是很好理解的，他们的操作也比较容易理解，直观：

主要有addAndGet、compareAndSet、getAndAdd(int delta)等等操作，AtomicReference主要用来更新引用！

对于AtomicBoolean来说，可能从使用标志位的角度来看，和voliatile Boolean区别不大，但是在做CAS操作的时候，区别就显示出来了：

volatile boolean aa;
Iif(aa){
       a=false;
 
       //.....XX
}
atomicBoolean.compareAndSet(true,false);

用法：

AtomicInteger atomic = new AtomicInteger(0);

atomic.incrementAndGet();

B2. 域更新器（field updater）：Integer、Long、Reference

如果查看Atomic计算器类的源码，会发现，内部都是一个volatile的变量，再加上unsafe类去使用CAS更新：

Uunsafe.compareAndSwapInt(this, valueOffset, expect, update);

域更新对象允许你不用包装一个类似AtomicInteger的对象，而是自己直接操作volatile类型的变量，并通过域更新器CAS操作变量，比如ConcurrentLinkedQueue的域更新器：

private transient volatile Node<E> head = new Node<E>(null, null);

private static final
        AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>
        tailUpdater =
        AtomicReferenceFieldUpdater.newUpdater
        (ConcurrentLinkedQueue.class, Node.class, "tail");

域更新器并不像计算器那样有构造函数，因为它是直接操作其他类的变量，因此需要以下3个参数：

1.操作的对象

2.操作的字段类型

3.操作的字段名

有了这三个参数，域更新器内部还是使用unsafe来进行CAS操作！使用域更新器有一定的缺陷，不能保证完全是原子的，因为如果volatile变量暴露出去的话，谁都可以操作，而不是通过域更新器操作，那就不能保证是原子的！

为什么要有域更新器？

即然他有这些缺陷，并且他能做的AtomicInteger都能做了，为什么还要域更新器？有一个理由是性能原因，下面是原话：对于频繁分配的，生命周期短暂的对象，比如队列里的Node，减少每个Node的AtomicReference的创建，对于减小插入操作的开销是非常有效地！（完全没看出来，我也搞不明白为什么要这个东西，哪位大虾解释下）

B3. 数组：Integer、Long、Reference

这些类可以保证数组里元素操作的原子性，需要多一个参数，那就是数组元素的位置！

B4. 复合（版本）变量

AtomicMarkableReference 类将单个布尔值与引用关联起来。例如，可以在数据结构内部使用此位，这意味着引用的对象在逻辑上已被删除。

boolean

isMarked()
返回该标记的当前值。

AtomicStampedReference 类将整数值与引用关联起来。例如，这可用于表示与更新系列对应的版本号

boolean

compareAndSet(V expectedReference, V newReference, int expectedStamp, int newStamp)
如果当前引用 == 预期引用，并且当前标志等于预期标志，则以原子方式将该引用和该标志的值设置为给定的更新值。

int

getStamp()
返回该标志的当前值。

C非阻塞算法

个线程的失败或挂起不应该影响其他线程的失败或挂起.这类算法称之为非阻塞(nonblocking)算法。

原来的同步类，比如HashTable，put、get方法都加了syn，但是他们同步的范围太广了，在相同功能的情况下，使用非阻塞算法比同步更好，要实现非阻塞，关键就是把原子的范围缩小到一个变量！

非阻塞栈

栈是最简单的数据结构，作为一种工具，它只有入/出栈两种操作，内部不管是数组或者链表，都是操作栈顶元素，因此可以将原子操作缩小到对栈顶元素的操作上来，内部只有一个栈顶指针head！

Push：将原来的栈顶cas替换为新的

Pop：将原来的栈顶设置为下一个元素

使用 Treiber 算法的非阻塞堆栈

public class ConcurrentStack<E> {
    AtomicReference<Node<E>> head = new AtomicReference<Node<E>>();
    public void push(E item) {
        Node<E> newHead = new Node<E>(item);
        Node<E> oldHead;
        do {
            oldHead = head.get();
            newHead.next = oldHead;
        } while (!head.compareAndSet(oldHead, newHead));
    }
    public E pop() {
        Node<E> oldHead;
        Node<E> newHead;
        do {
            oldHead = head.get();
            if (oldHead == null) 
                return null;
            newHead = oldHead.next;
        } while (!head.compareAndSet(oldHead,newHead));   //cas操作栈顶
        return oldHead.item;
    }
    static class Node<E> {
        final E item;
        Node<E> next;
        public Node(E item) { this.item = item; }
    }
}

非阻塞队列

入队

队列为了效率上的考虑，一般是用一个队首指针head和队尾指针tail来表示当前可以出队或者入队的位置，不管是内部是用数组或者链表，在更新对尾的时候都涉及到两个变量的更新（以链表结构的队尾为例）：1.更新队尾指针，tail 2.更新真正队尾元素的next为新的元素

这里有两个操作，很难归结到一个节点的原子操作，但是在状态上通过原子操作总能使其一致：（多一个dummy节点会相对方便很多，它将tail和head一开始就联系在一起，有入队的时候，head也能取到下一个元素，否则，他们两之间还得通过一些特殊操作关联起来）

图 1. 有两个元素，处在静止状态的队列


首先插入一个元素涉及两个指针更新，这两个更新都是通过 CAS 进行的：从队列当前的最后节点（C）链接到新节点，并把尾指针移动到新的最后一个节点（D）。如果第一步失败，那么队列的状态不变，插入线程会继续重试，直到成功。一旦操作成功，插入被当成生效，其他线程就可以看到修改。还需要把尾指针移动到新节点的位置上，但是这项工作可以看成是 “清理工作”，因为任何处在这种情况下的线程都可以判断出是否需要这种清理，也知道如何进行清理。

队列总是处于两种状态之一：正常状态（或称静止状态，图 1 和图 3）或中间状态（图 2）。在插入操作之前和第二个 CAS（D）成功之后，队列处在静止状态；在第一个 CAS（C）成功之后，队列处在中间状态。在静止状态时，尾指针指向的链接节点的 next 字段总为 null，而在中间状态时，这个字段为非 null。任何线程通过比较 tail.next 是否为 null，就可以判断出队列的状态，这是让线程可以帮助其他线程 “完成” 操作的关键。

图 2. 处在插入中间状态的队列，在新元素插入之后，尾指针更新之前

插入操作在插入新元素（A）之前，先检查队列是否处在中间状态，如清单 4 所示。如果是在中间状态，那么肯定有其他线程已经处在元素插入的中途，在步骤（C）和（D）之间。不必等候其他线程完成，当前线程就可以 “帮助” 它完成操作，把尾指针向前移动（B）。如果有必要，它还会继续检查尾指针并向前移动指针，直到队列处于静止状态，这时它就可以开始自己的插入了。

第一个 CAS（C）可能因为两个线程竞争访问队列当前的最后一个元素而失败；在这种情况下，没有发生修改，失去 CAS 的线程会重新装入尾指针并再次尝试。如果第二个 CAS（D）失败，插入线程不需要重试 —— 因为其他线程已经在步骤（B）中替它完成了这个操作！

图 3. 在尾指针更新后，队列重新处在静止状态


public class LinkedQueue <E> {

    private static class Node <E> {
        final E item;
        final AtomicReference<Node<E>> next;
        Node(E item, Node<E> next) {
            this.item = item;
            this.next = new AtomicReference<Node<E>>(next);
        }
    }
    private AtomicReference<Node<E>> head
        = new AtomicReference<Node<E>>(new Node<E>(null, null));
    private AtomicReference<Node<E>> tail = head;
    public boolean put(E item) {
        Node<E> newNode = new Node<E>(item, null);
        while (true) {
            Node<E> curTail = tail.get();
            Node<E> residue = curTail.next.get();
            if (curTail == tail.get()) {
                if (residue == null) /* A null表示稳定状态*/ {
                    if (curTail.next.compareAndSet(null, newNode)) /* C 设置next*/ 		{
		/* D 设置tail，失败没关系，反正有人帮它完成1*/ ;
                        tail.compareAndSet(curTail, newNode)
                        return true;
                    }
                } else {
                    tail.compareAndSet(curTail, residue) /* B 中间状态直接帮它完成*/;
                }
            }
        }
    }
}

出队

在上面入队的过程中，也可能正在出对，head不断地向后移动，在正常情况下，只要确保head能够cas指向下一个元素，但是，在初始化（head==tail）或者高并发（不断出队，head急速向后移动）的情况下，最终会head==tail，在这种情况下，有可能出于稳定状态，也可能出于中间状态：

public E poll() {
        for (;;) {
            Node<E> h = head;
            Node<E> t = tail;
            Node<E> first = h.getNext();
            if (h == head) {
                if (h == t) {
                    if (first == null)   //稳定态，队列为空
                        return null;
                    else					//中间状态，要先把tail设置好
                        casTail(t, first);
                } else if (casHead(h, first)) {	//head保证是原子操作
                    E item = first.getItem();
                    if (item != null) {
                        first.setItem(null);
                        return item;
                    }
                    // else skip over deleted item, continue loop,
                }
            }
        }
    }

总结

CAS操作是并发包的核心，前面已经说过，如果将Lock、CountDownLatch、FutureTask等最后都看成是依赖“状态”的操作，“状态”允许的情况下ok，否则挂起，那么对于“状态”的操作就需要同步的或者是原子的，因此原子操作就是这些类的核心，或者说前提条件！