锁是java并发的重要机制,它除了可以让临界区互斥执行以外,还可以向其它线程发送消息。
监视器锁
class MonitorExample { int a = 0; public synchronized void write() { //1 a++; //2 } //3 public synchronized void read() { //4 int i = a; //5 ... } //6 }下面使用happens-before来分析其执行过程。happens-before规则详见 http://blog.csdn.net/quanzhongzhao/article/details/45619135 。
1、程序顺序规则:1 happens before 2, 2 happens before 3; 4 happens before 5, 5 happens before 6。
2、监视器锁规则:3 happens before4。
3、happens-before规则的传递性:2 happens-before 5。
假如A线程先执行write()方法,即先获取监视器锁,然后B线程使用方法read()方法获取变量a的值。因为2 happens-before5,所以A线程在释放监视器锁之前对变量a的操作,在A释放监视器(按照监视器锁的语义,释放监视器 将本地内存中共享变量刷新到主内存)。所以随后B获取监视器锁(获取监视器会清空本地内存共享变量数据,并从主内存读取相应数据)后A对a的操作对B都是可见的。
JUC并发包中的锁
下面以ReenterantLock为例,说一下JUC并发包中的锁是如何实现的。
class ReentrantLockExample { int a = 0; ReentrantLock lock = new ReentrantLock(); public void write() { lock.lock(); //获取锁 try { a++; } finally { <span style="white-space:pre"> </span>lock.unlock(); //释放锁 <span style="white-space:pre"> </span>} } public void reade () { lock.lock(); //获取锁 try { int i = a; …… } finally { lock.unlock(); //释放锁 } } }ReentrantLock是一种互斥锁,但是是可重入的,即获得该锁的线程可以多次再获得该锁。ReentrantLock的实现依赖于其中的Sync类型变量。
而Sync是同步器AQS的子类,AQS即AbstractQueuedSynchronizer类。
AbstractQueuedSynchronizer使用一个整型
volatile变量来维护同步器的状态,这个volatile的变量state是锁内存语义实现的关键。
/** * The synchronization state. */ private volatile int state;
先看看ReentrantLock的构造函数吧
public ReentrantLock() { sync = new NonfairSync(); //默认使用非公平锁 } public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync() ; //若传入参数fair = true,则使用公平锁 }公平锁与非公平锁的区别主要在获得锁时的策略。后面再具体分析。下面分析一下我们使用lock.lock()方法和lock.unlock时候都发生了什么。
public void lock() { sync.lock(); }先来看看公平锁的lock.lock()方法
final void lock() { acquire(1); //直接调用acquire()方法。 }再来看看非公平锁的实现, 此处(lock方法)为公平锁与非公平锁的第一处不同。
final void lock() { if (compareAndSetState(0, 1)) //首先CAS判断同步器的状态state是否为0,为0表示当前锁可获得,获取该锁。 setExclusiveOwnerThread(Thread.currentThread()); //并设置当前线程为锁的拥有者。 else acquire(1); //若CAS失败,则同公平锁一样调用acquire()方法。 }acquire()方法由AbstractQueuedSynchronizer定义
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
宏观上解释一下acquire()方法
1、使用tryAcquire()方法尝试获取锁,获取成功则该方法直接返回true 表示获取成功 。该方法NonfairSync和FairSync都有实现。
2、尝试获取失败,则新生成一个节点(Node),其类型为互斥锁(Exclusive),存放当前线程。当前线程获取锁失败,则使用acquireQueued方法让等待队列中线程获取锁,
该方法是自旋的,即获取成功才能返回,保证了等待队列的非阻塞。且
该方法是非中断模式的,即不响应中断,但会记录中断状态。
3、acquireQueued方法调用过程中可能会有中断产生,有的话,执行完毕后在此处引发一个中断。
当多个线程尝试获取ReentrantLock锁时,只有一个线程能够独占该锁,其它线程则放入一个由Node组成等待队列中。其中每个Node持有一个线程,并维护节点的状态值。
static final class Node { static final Node SHARED = new Node(); //表示节点是共享节点 static final Node EXCLUSIVE = null; //独占节点 static final int CANCELLED = 1; static final int SIGNAL = -1; static final int CONDITION = -2; static final int PROPAGATE = -3; volatile int waitStatus; //节点状态值。为SIGNAL/CANCELLED/CONDITION/PROPAGATE。 volatile Node prev; //等待队列下一节点 volatile Node next; volatile Thread thread; //节点持有的线程。一个节点代表一个等待的线程。 Node nextWaiter; //用于waitStatus=CONDITION的节点。 Node() {} // Used to establish initial head or SHARED marker Node(Thread thread, Node mode) { // Used by addWaiter this.nextWaiter = mode; this.thread = thread; } Node(Thread thread, int waitStatus) { // Used by Condition this.waitStatus = waitStatus; this.thread = thread; } }
由上可知,节点的状态值也是一个volatile变量。
当waitStatus=SIGNAL时,表示该节点的后继节点正在阻塞状态,当前线程释放锁时或者当前线程状态变为CANCELLED,即取消获取锁时需要唤醒(unpark)后继节点。为了避免竞争,acquire方法首先指示我们需要一个Signal
acquireQueued(addWaiter(Node.EXCLUSIVE), arg) //即生成一个Node,其类型为EXCLUSIVE当waitStatus=CANCELLED时,表示由于超时或者中断导致该等待线程被取消,即线程不再获取锁。
当waitStatus=CONDITION时,表示节点在条件队列中。(目前不太理解这一状态)
当waitStatus=PROPAGATE时,用于共享锁。releaseShared应该传递给队列中的其它节点。用于doReleaseShared()方法中。
节点先说到这里,下面详细介绍acquire()方法。
protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); //AQS中tryAcquire()方法并未实现,也未使用抽象方法,考虑到AQS的两种功能(共享锁与互斥锁) } //避免子类实现抽象方法时需要实现两种功能。
//非公平锁版本 protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); //获取当前线程 int c = getState(); //获取同步器状态state if (c == 0) { if (compareAndSetState(0, acquires)) { //state=0,表示锁空闲,可以获取。使用CAS比较并置state的状态为1。 setExclusiveOwnerThread(current); //设置当前线程为锁的持有者。 return true; } } else if (current == getExclusiveOwnerThread()) { // state!=0,则表示锁已被某线程持有,则判断锁的持有线程 是否是 当前线程 int nextc = c + acquires; //锁的持有者为当前线程,则更新state的值。前面说过ReentrantLock可重入。 if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); //设置state的值为新值。 return true; } return false; //前面两种条件都不满足,则返回false,表示当前线程获取锁失败 }
//公平锁版本 protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); //获取当前线程 int c = getState(); //获取同步器状态 if (c == 0) { //当前同步器state=0,锁空闲,可获取 if (!hasQueuedPredecessors() && //判断是否有等待队列中是否有其它线程 compareAndSetState(0, acquires)) { //没有的话,则尝试获取锁 setExclusiveOwnerThread(current); //获取成功,设置当前线程为锁的持有者。 return true; } } else if (current == getExclusiveOwnerThread()) { //若state!=0,判断锁的持有线程是否是当前线程 int nextc = c + acquires; //是,则重入,并更新锁的状态值state if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; //表示获取锁成功 } return false; //获取失败 }当tryAcquire方法失败时,继而调用addWaiter()方法,新建节点持有当前线程,然后将其加入等待队列链表尾。
private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); //持有当前线程,接收参数mode=EXCLUSIVE 生成新节点, Node pred = tail; if (pred != null) { //判断等待队列是否为空 node.prev = pred; if (compareAndSetTail(pred, node)) { //队列非空,尝试CAS更新队尾。 pred.next = node; return node; //CAS成功,则返回该节点。 } } enq(node); //队列为空或者CAS失败 则以自旋的方式 加入队列。 return node; } private Node enq(final Node node) { for (;;) { //自旋,直到节点成功加入等待队列。 Node t = tail; if (t == null) { // Must initialize if (compareAndSetHead(new Node())) //队列为空,创建新的头结点。 tail = head; } else { node.prev = t; //非空,则CAS更改队列尾部 if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
final boolean acquireQueued(final Node node, int arg) { boolean failed = true; //表示是否成功获取锁 try { boolean interrupted = false; for (;;) { //自旋,直到 等待队列中某一线程 成功获取锁 final Node p = node.predecessor(); //获得当前节点前驱 if (p == head && tryAcquire(arg)) { //其前驱节点为队列头节点,则尝试获取锁。 setHead(node); //获取锁成功,设置当前节点为头结点, p.next = null; // help GC //此时原来的头结点 p 已经无效,设置为null有助于垃圾回收。 failed = false; return interrupted; //此过程中未产生中断,interrupted=false;所以acquire()方法不需要调用 selfInterrupt()方法。 } if (shouldParkAfterFailedAcquire(p, node) && //当前节点 前驱节点不是头节点,或者 是头结点但获取锁失 败调用该函数。 parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }继续展开 shouldParkAfterFailedAcquire
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; //获取前驱结点p的节点状态waitStatus if (ws == Node.SIGNAL) //前驱结点 waitStatus=SIGNAL,表示需要唤醒(unpark)其后继节点 return true; //返回,进而调用 parkAndCheckInterrupt() if (ws > 0) { //节点状态不是SIGNAL,大于0 ,只可能是CANCELLED ,表示前驱结点被取消,不再获取锁。 do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); //只要节点状态为CANCELLED,则继续向前搜索前驱节点,即为当前节点找到一个状态不为CANCELLED的前驱结点。 pred.next = node; //找到,将其设为当前节点的前驱节点。 } else { compareAndSetWaitStatus(pred, ws, Node.SIGNAL); //waitStatus must be 0 or PROPAGATE。 Indicate that we need a signal, but don't park yet } //Caller will need to retry to make sure it cannot acquire before parking return false; }
private final boolean parkAndCheckInterrupt() { //阻塞当前线程,并检查acquireQueued函数执行过程是是否发生中断 LockSupport.park(this); return Thread.interrupted(); }至此,lock.lock()方法执行完毕。
再看一下lock.unlock()方法。
public void unlock() { sync.release(1); }release()方法是在AQS中定义并实现的,子类并未实现。
public final boolean release(int arg) { if (tryRelease(arg)) { //尝试释放锁 Node h = head; if (h != null && h.waitStatus != 0) //释放成功判断头结点是否为空,并且判断其waitStatus状态 unparkSuccessor(h); //唤醒后继节点。 return true; } return false; //释放失败 } protected boolean tryRelease(int arg) { //AQS中tryRelease()方法并未实现,在子类Sync中实现。 throw new UnsupportedOperationException(); }tryRelease()方法在Sync中实现,并无公平锁与非公平锁之分。
protected final boolean tryRelease(int releases) { int c = getState() - releases; if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; if (c == 0) { free = true; setExclusiveOwnerThread(null); } setState(c); //释放锁的最后,写volatile变量state return free; }释放锁的最后写volatile变量state;在获取锁时首先读这个volatile变量。根据volatile的happens-before规则,释放锁的线程对volatile变量的写会刷新到主内存,获取锁的线程读会强制从主内存来读取该共享变量。即volatile变量保证了共享变量在线程之间的可见性。
最后说一下CAS方法,该方法时原子的,判断state值是否为expect,是则CAS更改state=update;
protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }CAS操作具有volatile 读和写的内存语义。编译器不会对volatile读与volatile读后面的任意内存操作重排序;编译器不会对volatile写与volatile写前面的任意内存操作重排序。组合这两个条件,意味着为了同时实现volatile读和volatile写的内存语义,编译器不能对CAS与CAS前面和后面的任意内存操作重排序。而具体实现的时候,CAS是依靠处理器指令级别的控制来实现原子操作。
现在对公平锁和非公平锁的内存语义做个总结:
1、公平锁和非公平锁释放时,最后都要写一个volatile变量state。
2、 公平锁获取时,首先会去读这个volatile变量。
3、非公平锁获取时,首先会用CAS更新这个volatile变量,这个操作同时具有volatile读和volatile写的内存语义。
JUC锁内存语义实现的总结
从本文对ReentrantLock的分析可以看出,锁释放-获取的内存语义的实现至少有下面两种方式:
1. 利用volatile变量的写-读所具有的内存语义。
2. 利用CAS所附带的volatile读和volatile写的内存语义。
1. 利用volatile变量的写-读所具有的内存语义。
2. 利用CAS所附带的volatile读和volatile写的内存语义。
由于java的CAS同时具有volatile 读和volatile写的内存语义,因此Java线程之间的通信现在有了下面四种方式:
1. A线程写volatile变量,随后B线程读这个volatile变量。
2. A线程写volatile变量,随后B线程用CAS更新这个volatile变量。
3. A线程用CAS更新一个volatile变量,随后B线程用CAS更新这个volatile变量。
4. A线程用CAS更新一个volatile变量,随后B线程读这个volatile变量。
Java的CAS会使用现代处理器上提供的高效机器级别原子指令,这些原子指令以原子方式对内存执行读-改-写操作,这是在多处理器中实现同步的关键(从本质上来说,能够支持原子性读-改-写指令的计算机器,是顺序计算图灵机的异步等价机器,因此任何现代的多处理器都会去支持某种能对内存执行原子性读-改-写操作的原子指令)。同时,volatile变量的读/写和CAS可以实现线程之间的通信。把这些特性整合在一起,就形成了整个concurrent包得以实现的基石。如果我们仔细分析concurrent包的源代码实现,会发现一个通用化的实现模式:
1. 首先,声明共享变量为volatile;
2. 然后,使用CAS的原子条件更新来实现线程之间的同步;
3. 同时,配合以volatile的读/写和CAS所具有的volatile读和写的内存语义来实现线程之间的通信。
AQS,非阻塞数据结构和原子变量类(java.util.concurrent.atomic包中的类),这些concurrent包中的基础类都是使用这种模式来实现的,而concurrent包中的高层类又是依赖于这些基础类来实现的。
本文锁的语义实现部分主要参考了程晓明<深入理解java内存模型>第六章锁的实现,其中最后仿宋字体部分为直接摘抄,其中AQS继承体系图来自
Java多线程系列--“JUC锁”03之 公平锁(一) 。如有错漏,多多指出啊。