第一节 synchronized关键字详解-偏向锁、轻量级锁、偏向锁、重量级锁、自旋、锁粗化、锁消除
第二节 AQS抽象队列同步器原理详解
ReentrantLock是一种基于AQS框架的应用实现，是JDK中的一种线程并发访问的同步手段，它的功能类似于synchronized，是一种互斥锁，可以保证线程安全。而且它具有比synchronized更多的特性，比如它支持手动加锁与解锁，支持加锁的公平性。
首先synchronized不具备的公平与非公平性，在ReentrantLock内部定义了一个Sync的内部类，该类继承AbstractQueuedSynchronized，对该抽象类的部分方法做了实现；ReentrantLock内部还定义了两个子类FairSync（公平锁）和NonfairSync（非公平锁），这两个类都继承自Sync，也就是间接继承了AbstractQueuedSynchronized，所以这一个ReentrantLock同时具备公平与非公平特性。
公平锁：
ReentrantLock lock = new ReentrantLock(true);
非公平锁：
ReentrantLock lock = new ReentrantLock(false);
默认情况下提供非公平锁，同时也可以在实例化的时候指定锁公平性
以下为源码部分：
#默认为非公平锁
public ReentrantLock() {
sync = new NonfairSync();
}
#可指定公平锁
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
公平和非公平的含义：
可重入指某个线程已经获得锁，可以再次获取锁而不会出现死锁，获取多少次锁就需要解锁多少次。
package com.xj;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class ReentrantLocks {
private static ReentrantLock lock = new ReentrantLock(true);
public static void reentrantLock(){
String threadName = Thread.currentThread().getName();
lock.lock();
log.info("Thread:{},第一次加锁",threadName);
lock.lock();
log.info("Thread:{},第二次加锁",threadName);
lock.unlock();
try{
Thread.sleep(1000);
}catch (Exception e){
}
log.info("Thread:{},第一次解锁",threadName);
lock.unlock();
log.info("Thread:{},第二次解锁",threadName);
}
public static void main(String[] args) {
Thread t0 = new Thread(new Runnable() {
@Override
public void run() {
reentrantLock();
}
},"t0");
t0.start();
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
reentrantLock();
}
},"t1");
t1.start();
}
}
打印结果：
20:14:36.353 [t0] INFO com.xj.ReentrantLocks - Thread:t0,第一次加锁
20:14:36.353 [t0] INFO com.xj.ReentrantLocks - Thread:t0,第二次加锁
20:14:37.366 [t0] INFO com.xj.ReentrantLocks - Thread:t0,第一次解锁
20:14:37.366 [t0] INFO com.xj.ReentrantLocks - Thread:t0,第二次解锁
20:14:37.366 [t1] INFO com.xj.ReentrantLocks - Thread:t1,第一次加锁
20:14:37.366 [t1] INFO com.xj.ReentrantLocks - Thread:t1,第二次加锁
20:14:38.372 [t1] INFO com.xj.ReentrantLocks - Thread:t1,第一次解锁
20:14:38.372 [t1] INFO com.xj.ReentrantLocks - Thread:t1,第二次解锁
package com.xj;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class ReentrantLock_Interrupt {
private static ReentrantLock lock = new ReentrantLock(true);
public static void reentrantLock() {
String threadName = Thread.currentThread().getName();
boolean flag = false;
lock.lock();
log.info("Thread:{},加锁成功!",threadName);
while(true){
if(Thread.currentThread().isInterrupted()){
log.info("Thread-Interrupted(线程被中断):{}",threadName);
break;
}
}
lock.unlock();
log.info("Thread:{},锁退出同步块",threadName);
}
public static void main(String[] args) {
//t0线程
Thread t0 = new Thread(new Runnable() {
@Override
public void run() {
String threadName = Thread.currentThread().getName();
try {
reentrantLock();
} catch (Exception e) {
log.info("Thread-InterruptedException:{}",threadName);
}
}
},"t0");
t0.start();
log.info("t0线程启动");
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
//t1线程
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
String threadName = Thread.currentThread().getName();
try {
reentrantLock();
} catch (Exception e) {
log.info("Thread-InterruptedException:{}",threadName);
}
}
},"t1");
t1.start();
log.info("t1线程启动");
try {
log.info("main线程休眠5S");
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.info("中断t0线程");
t0.interrupt();
t1.interrupt();
}
}
打印结果：
20:51:04.592 [main] INFO com.xj.ReentrantLock_Interrupt - t0线程启动
20:51:04.592 [t0] INFO com.xj.ReentrantLock_Interrupt - Thread:t0,加锁成功!
20:51:05.108 [main] INFO com.xj.ReentrantLock_Interrupt - t1线程启动
20:51:05.108 [main] INFO com.xj.ReentrantLock_Interrupt - main线程休眠5S
20:51:10.116 [main] INFO com.xj.ReentrantLock_Interrupt - 中断t0线程
20:51:10.116 [t0] INFO com.xj.ReentrantLock_Interrupt - Thread-Interrupted(线程被中断):t0
20:51:10.116 [t0] INFO com.xj.ReentrantLock_Interrupt - Thread:t0,锁退出同步块
20:51:10.116 [t1] INFO com.xj.ReentrantLock_Interrupt - Thread:t1,加锁成功!
20:51:10.116 [t1] INFO com.xj.ReentrantLock_Interrupt - Thread-Interrupted(线程被中断):t1
20:51:10.116 [t1] INFO com.xj.ReentrantLock_Interrupt - Thread:t1,锁退出同步块
package com.xj;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class ReentrantLock_Interrupt {
private static ReentrantLock lock = new ReentrantLock(true);
public static void reentrantLock() throws InterruptedException {
String threadName = Thread.currentThread().getName();
boolean flag = false;
lock.lockInterruptibly();
log.info("Thread:{},加锁成功!",threadName);
while(true){
if(Thread.currentThread().isInterrupted()){
log.info("Thread-Interrupted(线程被中断):{}",threadName);
break;
}
}
lock.unlock();
log.info("Thread:{},锁退出同步块",threadName);
}
public static void main(String[] args) {
//t0线程
Thread t0 = new Thread(new Runnable() {
@Override
public void run() {
String threadName = Thread.currentThread().getName();
try {
reentrantLock();
} catch (InterruptedException e) {
log.info("Thread-InterruptedException:{}",threadName);
}
}
},"t0");
t0.start();
log.info("t0线程启动");
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
//t1线程
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
String threadName = Thread.currentThread().getName();
try {
reentrantLock();
} catch (InterruptedException e) {
log.info("Thread-InterruptedException:{}",threadName);
}
}
},"t1");
t1.start();
log.info("t1线程启动");
try {
log.info("main线程休眠5S");
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.info("中断t1线程");
t1.interrupt();
}
}
打印结果：
21:00:31.220 [t0] INFO com.xj.ReentrantLock_Interrupt - Thread:t0,加锁成功!
21:00:31.220 [main] INFO com.xj.ReentrantLock_Interrupt - t0线程启动
21:00:31.735 [main] INFO com.xj.ReentrantLock_Interrupt - t1线程启动
21:00:31.735 [main] INFO com.xj.ReentrantLock_Interrupt - main线程休眠5S
21:00:36.751 [main] INFO com.xj.ReentrantLock_Interrupt - 中断t1线程
21:00:36.751 [t1] INFO com.xj.ReentrantLock_Interrupt - Thread-InterruptedException:t1
ReentrantLock#lock调用了sync.lock()，Sync中lock()是一个抽象函数，具体的实现在其子类NonfairSync和FairSync中。
public void lock() {
sync.lock();
}
抽象类Sync：
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860L;
/**
* Performs {@link Lock#lock}. The main reason for subclassing
* is to allow fast path for nonfair version.
*/
abstract void lock();
//省略
}
ReentrantLock.NonfairSync#lock实现了Sync中的抽象方法lock，其非公平性体现在，一上来就抢锁设置state状态compareAndSetState(0,1)，如果抢锁成功就设置当前线程为持有独占锁的线程setExclusiveOwnerThread(Thread.currentThread())。
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
/**
* Performs lock. Try immediate barge, backing up to normal
* acquire on failure.
*/
final void lock() {
//非公平，直接CAS进行修改state状态，进行抢锁
if (compareAndSetState(0, 1))
//抢成功，设置当前独占锁的线程为当前线程
setExclusiveOwnerThread(Thread.currentThread());
else
//没有抢到，会再次抢一次，没有抢到则进入同步队列
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
如果一开始抢锁失败则进入AQS的模板方法acquire(1)，acquire(1)中还会再进行一次抢锁tryAcquire
，抢锁还是失败才进入AQS队列操作acquireQueued(addWaiter(Node.EXCLUSIVE), arg)。
public final void acquire(int arg) {
//没有抢到锁，则进入等待队列
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
//自己中断自己，设置中断标记
selfInterrupt();
}
加锁的通用模板acquire()已经在AQS中实现，子类只需要实现tryAcquire即可。NonfairSync#tryAcquire调用了父类的Sync#nonfairTryAcquire。
具体逻辑如下：
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
//c=0,此时锁空着，直接尝试抢锁
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
//如果当前持锁线程是当前线程(可重入)
else if (current == getExclusiveOwnerThread()) {
//
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
FairSync和NonfairSync的主要区别在于，FairSync不会一上来就抢锁，而是先判断队列中是否有其他线程在等待锁，没有再抢锁。代码的模板逻辑和NonfairSync类似，区别在于FairSync中tryAcquire
的实现。
public final void acquire(int arg) {
//没有抢到锁，则进入等待队列
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
//自己中断自己，设置中断标记
selfInterrupt();
}
/**
* Fair version of tryAcquire. Don't grant access unless
* recursive call or no waiters or is first.
*/
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
//判断队列中是否有线程在等待锁，没有则抢锁
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
//如果当前线程已经持有该锁，可重入
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
}
hasQueuedPredecessors（）源码如下：
public final boolean hasQueuedPredecessors() {
// The correctness of this depends on head being initialized
// before tail and on head.next being accurate if the current
// thread is first in queue.
//指针t指向队尾
Node t = tail; // Read fields in reverse initialization order
//指针h指向队头
Node h = head;
Node s;
//队头不等于队尾且队头的后继节点为null或者是后继节点的线程非当前线程
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}
acquireQueued（）源码如下：
final boolean acquireQueued(final Node node, int arg) {
//失败标记
boolean failed = true;
try {
//中断标记
boolean interrupted = false;
//自旋
for (;;) {
//获取node节点的前驱节点
final Node p = node.predecessor();
//如果当前节点的前驱节点为队头，且当前线程再次尝试获取锁
if (p == head && tryAcquire(arg)) {
//当前线程获取到了锁，将当前节点设置为头节点
setHead(node);
//前驱节点从队列中移除
p.next = null; // help GC
//没有发现错误
failed = false;
//返回此时的中断标记
return interrupted;
}
//如果条件不满足，则进入以下的逻辑
//shouldParkAfterFailedAcquire(p, node)返回true只能是前驱节点的信号量状态为SIGNAL,不为SIGNAL且不为CANCELLED，会先执行compareAndSetWaitStatus修改信号量为SIGNAL状态
//
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
//说明是中断唤醒的线程，设置中断标记true
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
//将当前节点设置为头节点
private void setHead(Node node) {
head = node;
//清空当前节点thread和prev内容
node.thread = null;
node.prev = null;
}
//检查和更新获取失败的节点的状态。
//如果线程阻塞，返回true
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
//获取前驱节点的信号量状态
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;//如果为SIGNAL（-1），返回true
if (ws > 0) {//>0的情况，只能是CANCELLED状态，需要被跳过
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
//找前驱节点的前驱节点，并判断其的信号量状态
//不为>0,退出循环
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
//将找出来的前驱节点的后继节点指针指向当前节点
pred.next = node;
} else {
/*
* 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.
*/
//waitStatus必须为0或PROPAGATE。
//CAS修改WaitStatus为SIGNAL
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
//线程驻留，然后返回中断标记
private final boolean parkAndCheckInterrupt() {
//线程驻留
LockSupport.park(this);
//之所以需要返回中断标记，是因为线程如果被中断会被唤醒，所以需要在这里返回状态，并清空中断标记
return Thread.interrupted();
}
//自己中断自己，设置中断标记
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
lock中断是通过自己设置中断标记以及业务自己在程序代码里面进行退出线程，从而达到中断的效果。lockInterruptibly是直接抛出中断异常，其调用了AQS的模板方法acquireInterruptibly。
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
public final void acquireInterruptibly(int arg)
throws InterruptedException {
//判断中断标记，并清空
//true抛出中断异常
if (Thread.interrupted())
throw new InterruptedException();
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}
tryAcquire和上面的源码分析类似，而doAcquireInterruptibly也类似，没有了中断标记设置相关的逻辑，而是抛出了中断异常。
private void doAcquireInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.EXCLUSIVE);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
tryLock尝试获取锁，调用的是sync.nonfairTryAcquire(1)，使用的是非公平获取锁，不会涉及到AQS队列操作，获取锁成功返回true，失败返回false。
public boolean tryLock() {
return sync.nonfairTryAcquire(1);
}
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
//c=0,此时锁空着，直接尝试抢锁
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
//如果当前持锁线程是当前线程(可重入)
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
tryLock还有一个重载方法，可传入一个超时时长timeout和一个时间单位TimeUnit，超时时长会被转为纳秒级。
public boolean tryLock(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
//判断中断标记并清空
if (Thread.interrupted())
//抛出中断异常
throw new InterruptedException();
return tryAcquire(arg) ||
doAcquireNanos(arg, nanosTimeout);
}
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
//判断超时时间
if (nanosTimeout <= 0L)
return false;
//截止时间=系统当前时间+超时时间
final long deadline = System.nanoTime() + nanosTimeout;
//进入同步等待队列
final Node node = addWaiter(Node.EXCLUSIVE);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return true;
}
//获取超时时间
nanosTimeout = deadline - System.nanoTime();
//<0l超时时间已到,返回false，代表获取锁超时
if (nanosTimeout <= 0L)
return false;
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
//线程停留超时时间
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
释放锁的过程没有区分公平性，调用的是AQS的模板方法release()
public void unlock() {
sync.release(1);
}
其基本逻辑如下：
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
释放锁的代码ReentrantLock.Sync#tryRelease基本逻辑如下：
protected final boolean tryRelease(int releases) {
//计算释放锁之后的state值
int c = getState() - releases;
//不是当前线程，不能unLock 抛异常
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
//c = 0,证明没有线程持有锁了,将独占线程字段内容清空
free = true;
setExclusiveOwnerThread(null);
}
//线程安全,此为排它锁，只有当前线程会走到这修改state
setState(c);
return free;
}
注：如果是当前线程多次重入，releases=1是不能完全释放锁的，free=false，也不会唤醒后继节点。
/**
* Wakes up node's successor, if one exists.
* 唤醒后继节点
*
* @param node the node
*/
private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/
//当前节点信号量状态<0 为负数
int ws = node.waitStatus;
if (ws < 0)
//CAS算法设置WaitStatus为0
compareAndSetWaitStatus(node, ws, 0);
/*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
//指针s指向node的后继节点
Node s = node.next;
//后继节点为空或者是信号量状态>0(CANCELLED)
if (s == null || s.waitStatus > 0) {
//后继节点置空
s = null;
//查找符合条件的节点，从尾部向后遍历查找
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
//后继节点不为空
if (s != null)
//解除驻留
LockSupport.unpark(s.thread);
}
CAS操作对象基于unsafe类，其无法实例化，只能通过反射来获取Unsafe，以下是获取unsafe实例代码：
public class UnsafeInstance {
public static Unsafe reflectGetUnsafe() {
try {
Field field = Unsafe.class.getDeclaredField("theUnsafe");
field.setAccessible(true);
return (Unsafe) field.get(null);
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
}
CAS案例：
public class CompareAndSwap {
/**
* 记录当前加锁状态
*/
volatile int state = 0;
private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5, new Runnable() {
@Override
public void run() {
log.info("所有线程都准备好了.....");
}
});
private static CompareAndSwap cas = new CompareAndSwap();
public static void main(String[] args) {
new Thread(new WorkerThread(),"t0").start();
new Thread(new WorkerThread(),"t1").start();
new Thread(new WorkerThread(),"t2").start();
new Thread(new WorkerThread(),"t3").start();
new Thread(new WorkerThread(),"t4").start();
}
static class WorkerThread implements Runnable{
@Override
public void run() {
log.info("线程{}到达预定点,准备开始抢state:",Thread.currentThread().getName());
try {
cyclicBarrier.await();
if(cas.compareAndSwapState(0,1)){
log.info("线程{},抢到锁!",Thread.currentThread().getName());
}else{
log.info("线程{},抢锁失败!",Thread.currentThread().getName());
}
} catch (InterruptedException| BrokenBarrierException e) {
e.printStackTrace();
}
}
}
/**
* 原子操作
* @param oldValue线程工作内存当中的值
* @param newValue:要替换的新值
* @return
*/
public final boolean compareAndSwapState(int oldValue,int newValue){
return unsafe.compareAndSwapInt(this,stateOffset,oldValue,newValue);
}
private static final Unsafe unsafe = UnsafeInstance.reflectGetUnsafe();
private static final long stateOffset;
static {
try {
stateOffset = unsafe.objectFieldOffset(CompareAndSwap.class.getDeclaredField("state"));
} catch (Exception e) {
e.printStackTrace();
throw new Error();
}
}
}
ABA问题在于cas在修改变量的时候，无法记录变量的状态，比如修改的次数，否修改过这个变量。这样就很容易在一个线程将A修改成B时，另一个线程又会把B修改成A,造成CAS多次执行的问题。
例如主线程操作账户余额过程中（进账100元）需要耗时一段时间，这段时间中干扰线程先进账100元，此时账户余额200，然后出账100元，此时账户余额100元，最终主线程得到账户余额为200，也就是金额出现了200->100->200情况
public class ABAProblem {
static AtomicInteger atomicInteger = new AtomicInteger(100);
public static void main(String[] args) {
Thread main = new Thread(new Runnable() {
@Override
public void run() {
int account = atomicInteger.get();
log.info(Thread.currentThread().getName()+"——修改前账户余额:"+account);
try {
log.info(Thread.currentThread().getName()+"——账户其它操作，需要一段时间");
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
if(atomicInteger.compareAndSet(account,account + 100)){
log.info(Thread.currentThread().getName()+"——CAS进账100元:"+atomicInteger.get());
}else{
log.info(Thread.currentThread().getName()+"——CAS修改账户余额失败");
}
}
},"主线程");
Thread other = new Thread(new Runnable() {
@Override
public void run() {
atomicInteger.compareAndSet(atomicInteger.get(),atomicInteger.get() + 100);// 1+1 = 2;
log.info(Thread.currentThread().getName()+"——CAS进账100元:"+atomicInteger.get());
atomicInteger.compareAndSet(atomicInteger.get(),atomicInteger.get() - 100);// atomic-1 = 2-1;
log.info(Thread.currentThread().getName()+"--CAS出账100元:"+atomicInteger.get());
}
},"干扰线程");
main.start();
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
other.start();
}
}
打印结果：
11:07:29.461 [主线程] INFO com.xj.ABAProblem - 主线程——修改前账户余额:100
11:07:29.463 [主线程] INFO com.xj.ABAProblem - 主线程——账户其它操作，需要一段时间
11:07:29.970 [干扰线程] INFO com.xj.ABAProblem - 干扰线程——CAS进账100元:200
11:07:29.970 [干扰线程] INFO com.xj.ABAProblem - 干扰线程--CAS出账100元:100
11:07:32.476 [主线程] INFO com.xj.ABAProblem - 主线程——CAS进账100元:200
通过AtomicStampReference类来解决该问题，其是在CAS的基础上增加了一个标记stamp，使用这个标记可以用来觉察数据是否发生变化，给数据带上了一种实效性的检验。
public class AtomicStampedRerenceExample {
//初始化AtomicStampedReference
private static AtomicStampedReference<Integer> atomicStampedRef =
new AtomicStampedReference<>(100, 0);
public static void main(String[] args){
//主线程
Thread main = new Thread(() -> {
//获取当前标识
int version = atomicStampedRef.getStamp();
System.out.println(Thread.currentThread().getName()+ ":version="+version + ",初始值account = " + atomicStampedRef.getReference());
try {
Thread.sleep(3000); //等待3秒 ，以便让干扰线程执行
} catch (InterruptedException e) {
e.printStackTrace();
}
boolean isCASSuccess = atomicStampedRef.compareAndSet(atomicStampedRef.getReference(),atomicStampedRef.getReference() + 100,version,version +1); //此时expectedReference未发生改变，但是stamp已经被修改了,所以CAS失败
System.out.println(Thread.currentThread().getName() + ":version="+version + ",CAS操作结果: " + isCASSuccess);
},"主线程");
//干扰线程
Thread other = new Thread(() -> {
//获取当前标识
int version = atomicStampedRef.getStamp();
atomicStampedRef.compareAndSet(atomicStampedRef.getReference(),atomicStampedRef.getReference() + 100,version,version+1);
System.out.println(Thread.currentThread().getName() + ":version="+atomicStampedRef.getStamp() +",[INCR],值 a= "+ atomicStampedRef.getReference());
//再次获取当前标识
version = atomicStampedRef.getStamp();
atomicStampedRef.compareAndSet(atomicStampedRef.getReference(),atomicStampedRef.getReference() - 100,version,version+1);
System.out.println(Thread.currentThread().getName() + ":version="+atomicStampedRef.getStamp() +",[DECR],值 a= "+ atomicStampedRef.getReference());
},"干扰线程");
main.start();
LockSupport.parkNanos(1000000);
other.start();
}
打印结果：
主线程:version=0,初始值account = 100
干扰线程:version=1,[INCR],值 a= 200
干扰线程:version=2,[DECR],值 a= 100
主线程:version=0,将账户余额增加100的CAS操作结果: false
LockSupport是concurrent包中一个工具类，不支持构造，提供了很多static方法，常见的比如park(),unpark()等。
public class LockSupportExample {
public static void main(String[] args) {
Thread t0 = new Thread(new Runnable() {
@Override
public void run() {
Thread current = Thread.currentThread();
log.info("线程{},开始执行!",current.getName());
for(;;){//spin 自旋
log.info("准备park当前线程:{}....",current.getName());
LockSupport.park();
//此处需要执行Thread.interrupted()，用于清空中断标记，否则将无法再次park
if(Thread.interrupted()){
log.info(current.getName()+"线程被中断唤醒");
}
log.info("当前线程{}已经被唤醒....",current.getName());
}
}
},"t0");
t0.start();
try {
Thread.sleep(5000);
log.info("准备唤醒{}线程!",t0.getName());
//unpark解除t0线程驻留
LockSupport.unpark(t0);
Thread.sleep(2000);
//中断唤醒t0线程
t0.interrupt();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
打印结果：
16:47:56.534 [t0] INFO com.xj.LockSupportExample - 线程t0,开始执行!
16:47:56.537 [t0] INFO com.xj.LockSupportExample - 准备park当前线程:t0....
16:48:01.532 [main] INFO com.xj.LockSupportExample - 准备唤醒t0线程!
16:48:01.532 [t0] INFO com.xj.LockSupportExample - 当前线程t0已经被唤醒....
16:48:01.532 [t0] INFO com.xj.LockSupportExample - 准备park当前线程:t0....
16:48:03.547 [t0] INFO com.xj.LockSupportExample - t0线程被中断唤醒
16:48:03.547 [t0] INFO com.xj.LockSupportExample - 当前线程t0已经被唤醒....
16:48:03.547 [t0] INFO com.xj.LockSupportExample - 准备park当前线程:t0....
本节主要介绍的是ReentrantLock相关特性与源码分析，以及关于实现Lock锁的CAS与LockSupport的核心原理。