Java并发锁机制 博客分类： java Java并发锁多线程读写锁
在进行多线程编程时，经常遇到多个线程同时对一个变量进行修改的问题。这时候为了保证不出现意想不到的结果，需要为这些变量加锁，以保证同一时刻只有一个线程能够修改。
在Java语言中，为了解决这种同步互斥的访问，有两种方法：synchronized和Lock.
1,synchronized
synchronized是Java语言中的一个关键词，是Java语言内置的特性。
通过使用synchronized来修饰一个方法或者一个代码块，来获得相应对象的锁，使这个方法或者代码块实现同步互斥访问。
synchronized方法
当一个线程执行某个对象的synchronized方法时，其他线程不能访问该对象的所有synchronized方法，因为一个对象只有一把锁。但是可以访问非synchronized方法。
另外，每个类也有一个锁，用了控制对static 成员变量的访问。
当一个线程访问一个对象的非static synchronized方法时，另一个线程可以访问该对象所属类的static synchronized方法。因为他们分别占用的是对象锁和类锁，不存在同步互斥问题。
synchronized代码块
当synchronized修饰代码块时，是如下格式使用，即会指定一个对象或者某个属性，需要获取该对象或属性的锁，才能执行代码块：
synchronized(obj) {
}
synchronized代码块比synchronized方法更加灵活，
当一个线程获得锁，执行到synchronized修饰的代码时，其他的线程只能是一直等待状态。当前已获取锁的线程只有在以下情况下释放锁：
a, 执行完现在的代码块，线程释放对锁的占用；
b, 执行时出现异常，JVM会让线程自动释放该锁；
c, 线程调用wait()从而进入等待状态，自动释放锁。
因此synchronized存在如下一些缺陷：
a，未获取锁的线程只能一直等待当前的线程执行结束，才能获得锁。无法选择中断，如果当前的线程有一些耗时的操作的话，会很浪费资源。
b，通过synchronized来实现同步互斥的粒度太大。synchronized修饰的方法或代码块，实际上并不是方法或代码块中的所有代码都需要同步互斥访问。
c，对资源的操作分为读操作和写操作。两个读操作并不会出现并发互斥问题，如果使用synchronized的话，即使是两个读操作，也会在同一时刻只能有一个读操作能执行，其他的线程执行读操作会被阻塞。
d，我们无法得知线程是否已经成功获得锁。
由于存在以上这些问题，因此Java又引进了Lock机制。
2,Lock
Lock是一个Java接口，基于JDK层面的实现，来实现同步互斥访问。Lock的实现机制是以对volatile变量的读/写和CAS来实现多线程同步互斥访问的。
volatile：
由于CPU执行代码时，为了提高效率，会维护一个高速缓存。使用Volatile修饰的变量，都会在执行代码时强制从内存中读取最新变量值，而不是使用高速缓存的值。Volatile还会禁止指令重排。因此保证了两个特性：1，可见性，2有序性。
CAS： Compare and Swap(比较并交换)
CAS针对内存的操作：内存值V，旧的预期值A，要修改的新值B，当且仅当旧的预期值A与内存值V相同时，将内存值V修改为新值B，否则什么都不做。
借助CAS完成这个过程，是通过JNI调用C语言操作CPU底层指令完成的。这样就能保证这些内存操作的原子性。
我们分析一个使用Lock加锁的实例来查看，Lock是如何保证多线程的同步互斥访问的。
public Lock lk = new ReentrantLock();
public void testlock(){
lk.lock();
try {
//需要互斥访问的具体操作代码
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally{
lk.unlock();
}
}
Lock的使用，是通过调用lk.lock()和lk.unlock()来完成同步互斥访问的。
具体的lock()方法具体执行流程如下，以使用非公平锁为例说明，公平锁的流程稍有区别：
获取锁的过程即是对AQS中state状态进行修改的过程。state字段使用volatile修饰，使用CAS的方法compareAndSetState进行修改。如果state字段的值为0时，表示空闲，可以获取锁。大于0时，表示已经获得锁。当锁的模式为独占锁时，表示锁的重入次数，当锁的模式为共享锁时，表示锁当前共享的线程数。
3,Lock相关的类分析
了解了获取锁的大概流程后，我们来具体看看与锁相关的类。
Lock
Lock是一个接口，有以下几种方法：
lock(): 获取锁；
lockInterruptibly(): 获取锁，如果获取失败在等待锁时，能够响应中断，抛出中断异常。
trylock(): 获取锁，并且立即返回，获取锁成功则返回True，失败则返回False.
trylock(long,TimeUnit):与trylock类似，只是获取锁失败后则继续等待指定的时间。并且可以响应中断。
Condition
Condition：接口，提供了一些方法，以达到对锁更精确的控制。
await(): 线程等待状态，直到收到信号或者被中断。
awaitUninterruptibly():线程等待状态，直到收到信号。
awaitNanos(long):线程等待状态，直到收到信号或者被中断，或者到达指定时间为止。
await(long,TimeUnit):线程等待状态，直到收到信号或者被中断，或者到达指定时间为止。
signal(): 唤醒一个等待线程。
signalAll():唤醒所有等待线程。
AbstractOwnableSynchronizer：抽象类，由一个线程独占的同步器。
AbstractQueuedSynchronizer(AQS)：
继承AbstractOwnableSynchronizer的抽象类，维护一个等待获取锁的线程CLH队列。包含两个内部类： Node和ConditionObject.
AbstractQueuedSynchronizer-Node类： 等待队列的Node类。
Node分为两种模式：独占锁和共享锁模式。
Node分为5种状态： SIGNAL = -1 表示当前节点的后继节点包含的线程需要运行，需要unparking;
CANCELLED = 1 表示当前节点由于过期或者中断被取消了；
CONDITION = -2 表示当前节点在等待condition,即在Condition队列中;
PROPAGATE = -3 表示后续的acquireShared应该执行
0：表示当前节点在同步队列中，等待锁。
AbstractQueuedSynchronizer-ConditionObject类：Condition接口的实现类。
signal():移除一个等待时间最长的线程，即Condition等待队列的第一个线程，把这个线程从condition队列一道同步队列中，即把Node状态由CONDITION(-2)修改为0。
signalAll():同signal一样，移除Condition队列中的所有线程。
ReentrantLock
ReentrantLock是一个可重入锁，Lock接口的实现类。默认创建的ReentrantLock是非公平锁，ReentrantLock(boolean) 支持创建公平锁和非公平锁。包含一个抽象内部类Sync(继承AbstractQueuedSynchronizer)以及其两个子类FairSync和NonfairSync，分别实现公平锁和非公平锁。
FairSync类和NonfairSync类都只包含lock()和tryAcquire()。这两个方法实现的不同，展示了公平锁和非公平锁的不同。
FairSync的lock方法直接调用acquire(1)，AQS的acquire方法如下，即先调用tryAcquire(arg)和addWaiter(),acquireQueued，然后根据条件调用selfinterrupt方法。
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
FairSync重写的tryAcquire方法，先判断是否队列中有前置节点，没有前置节点才会尝试获取锁，这样就保证了多个线程获取锁的公平性。获取锁失败则返回False，然后让acquire方法继续完成加入独占锁等待队列。
NonfairSync的lock方法则是首先抢占式的获取锁，直接调用compareAndSetState方法，失败之后才会调用acquire(1).
NonfairSync重写的tryAcquire方法,尝试获取锁，获取失败加入等待队列。
tryAcquire方法中，都会判断当前线程是否已经获取锁的线程。如果是的话，则会更新state值为state+1，此时就是完成重入锁的过程。
ReentrantReadWriteLock
同时，JUC包还提供了ReadLock 和 WriteLock.
ReadLock和WriteLock都作为ReentrantReadWriteLock的内部类，实现了Lock接口。ReentrantReadWriteLock也包含一个抽象类Sync继承自AbstractQueuedSynchronizer，和两个子类FairSync和NonfairSync。在AQS类中，state的值代表锁的状态，在Sync类中，对state进行了如下定义：高16位代表读锁的数量，低16位代表写锁的重入次数。如下图所示：
读锁和写锁的工作过程：
通过ReentrantReadWriteLock能直接获得读锁和写锁，可传入参数决定是公平锁或非公平锁。
public ReentrantReadWriteLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
readerLock = new ReadLock(this);
writerLock = new WriteLock(this);
}
读锁的获取过程：
lock方法直接调用AQS的acquireShared(1),即以共享模式获取锁，然后调用Sync重写的tryAcquireShared(1)，进行判断，如果是其他的线程已经持有独占锁，则直接返回-1,获取锁失败，进入共享锁等待队列； 否则获取锁成功。
protected final int tryAcquireShared(int unused) {
/*
* Walkthrough:
* 1. If write lock held by another thread, fail.
* 2. Otherwise, this thread is eligible for
* lock wrt state, so ask if it should block
* because of queue policy. If not, try
* to grant by CASing state and updating count.
* Note that step does not check for reentrant
* acquires, which is postponed to full version
* to avoid having to check hold count in
* the more typical non-reentrant case.
* 3. If step 2 fails either because thread
* apparently not eligible or CAS fails or count
* saturated, chain to version with full retry loop.
*/
Thread current = Thread.currentThread();
int c = getState();
//如果是其他的线程已经持有独占锁，则直接返回-1.
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
//获得共享锁的数量
int r = sharedCount(c);
//如果不需要阻塞读锁，并且读锁数量没达到最大值，并且成功更新读锁的数量(在高16位加1);
、 //判断是否需要阻塞读锁，readerShouldBlock(),是公平锁和非公平锁的区别所在。
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
//r=0，表示当前线程就是第一个读锁。
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
//firstReader就是当前线程，则是当前线程重入了，更新firstReaderHoldCount.
firstReaderHoldCount++;
} else {
//当前线程和第一个线程不同，记录当前线程读锁+1.
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
cachedHoldCounter = rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
}
return 1;
}
//否则，循环尝试
return fullTryAcquireShared(current);
}
公平锁和非公平锁的区别
在调用readerShouldBlock时体现： 非公平锁只需要判断第一个线程节点是否是独占模式；公平锁则是判断队列中是否有前置线程在排队等待。
在调用writeShouldBlack时体现：非公平锁直接返回false,不需要阻塞；公平锁则是判断队列中是否有前置线程在排队等待。
写锁的获取过程：
直接调用AQS的acquire(1)，然后调用Sync重写的tryAcquire(1)，
protected final boolean tryAcquire(int acquires) {
/*
* Walkthrough:
* 1. If read count nonzero or write count nonzero
* and owner is a different thread, fail.
* 2. If count would saturate, fail. (This can only
* happen if count is already nonzero.)
* 3. Otherwise, this thread is eligible for lock if
* it is either a reentrant acquire or
* queue policy allows it. If so, update state
* and set owner.
*/
Thread current = Thread.currentThread();
int c = getState();
int w = exclusiveCount(c);
//c!=0代表当前已经有锁(读锁或者写锁)。
if (c != 0) {
// (Note: if c != 0 and w == 0 then shared count != 0)
//如果写锁为0，或者当前线程不是独占线程，
//即代表当前有锁，但不符合重入条件，返回false。
if (w == 0 || current != getExclusiveOwnerThread())
return false;
//写锁数量超过最大数；
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// Reentrant acquire
//设置当前写锁重入数
setState(c + acquires);
return true;
}
//当前还没有锁的处理过程：
//写锁应该阻塞或者直接修改标志位获取锁的操作(CAS)失败，返回false
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
return false;
//成功获取锁，则设置当前线程为独占线程;
setExclusiveOwnerThread(current);
return true;
}
读锁和写锁的释放unlock，操作流程也跟lock类似，分别调用releaseShared(1)->tryReleaseShared(1)和release(1)->tryRelease(1)。protected final boolean tryRelease(int releases) {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
int nextc = getState() - releases;
boolean free = exclusiveCount(nextc) == 0;
//只有当写锁的数量为0时，才把当前独占线程置为null
if (free)
setExclusiveOwnerThread(null);
setState(nextc);
return free;
}
当把写锁的数量降为0，才能返回true，否则一直返回false。protected final boolean tryReleaseShared(int unused) {
Thread current = Thread.currentThread();
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
if (firstReaderHoldCount == 1)
firstReader = null;
else
firstReaderHoldCount--;
} else {
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
rh = readHolds.get();
int count = rh.count;
if (count <= 1) {
readHolds.remove();
if (count <= 0)
throw unmatchedUnlockException();
}
--rh.count;
}
for (;;) {
int c = getState();
int nextc = c - SHARED_UNIT;
if (compareAndSetState(c, nextc))
// Releasing the read lock has no effect on readers,
// but it may allow waiting writers to proceed if
// both read and write locks are now free.
return nextc == 0;
}
}
读锁的释放，只有把读锁的数量降为0，才返回True，否则一直返回false。public static void main(String[] args) {
// TODO Auto-generated method stub
System.out.println("main thread start");
LockSupport.park();
System.out.println("main thread park");
}
//运行结果：
main thread start
System.out.println("main thread park");//这句代码并没有执行，而程序也没有结束，只是阻塞了。
//因为默认的permit为0，此时调用LockSupport.park(),
//需要等到其他线程把permit修改为1之后才会继续往下执行.
public static void main(String[] args) {
// TODO Auto-generated method stub
System.out.println("main thread start");
LockSupport.unpark(Thread.currentThread());
LockSupport.park();
System.out.println("main thread park");
}
//先调用LockSupport.unpark(Thread.currentThread())，则程序并没有阻塞，能正常运行结束。
public static void main(String[] args) {
// TODO Auto-generated method stub
Producer p1 = new Producer(Thread.currentThread());
Thread t1 = new Thread(p1);
t1.start();
System.out.println("main thread start");
LockSupport.park();
System.out.println("main thread park");
}
public class Producer implements Runnable {
private Thread mainThread;
public Producer(Thread mainThread) {
super();
this.mainThread = mainThread;
}
public void run() {
// TODO Auto-generated method stub
System.out.println("start thread");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
//在此调用主线程的unpark。则能够唤醒主线程，使之继续往下执行。
LockSupport.unpark(this.mainThread);
System.out.println("end thread");
}
}
public static void main(String[] args) {
// TODO Auto-generated method stub
Producer p1 = new Producer(Thread.currentThread());
Thread t1 = new Thread(p1);
t1.start();
System.out.println("main thread start");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("main thread start to interrupt subthread");
t1.interrupt();
System.out.println("main thread end");
}
public class Producer implements Runnable {
private Thread mainThread;
public Producer(Thread mainThread) {
super();
this.mainThread = mainThread;
}
public void run() {
// TODO Auto-generated method stub
System.out.println("start subthread");
LockSupport.park();
System.out.println("end subthread, subthread interrupted status:" + Thread.currentThread().isInterrupted());
}
}
上一篇： 简单谈谈java中匿名内部类构造函数
下一篇： Java代码性能优化的35个方法总结