ReentrantLock的使用及原理
定义
public class ReentrantLock implements Lock, java.io.Serializable {
private final Sync sync;
abstract static class Sync extends AbstractQueuedSynchronizer {
abstract void lock();
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (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;
}
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);
return free;
}
}
public ReentrantLock() {
sync = new NonfairSync();
}
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
public void lock() {
sync.lock();
}
public void unlock() { sync.release(1);}
public Condition newCondition() {
return sync.newCondition();
}
...
}
|
使用方式
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
lock.lock();
try {
while(条件判断表达式) {
condition.wait();
}
} finally {
lock.unlock();
}
|
在深入理解ReentrantLock的实现原理之前，我们先了解一下java同步器。推荐博客：深入浅出java同步器
和Aqs的关系
非公平锁实现
在非公平锁中，每当线程执行lock方法时，都尝试利用CAS把state从0设置为1。
那么Doug lea是如何实现锁的非公平性呢？
我们假设这样一个场景：
- 持有锁的线程A正在running，队列中有线程BCDEF被挂起并等待被唤醒；
- 在某一个时间点，线程A执行unlock，唤醒线程B；
- 同时线程G执行lock，这个时候会发生什么？线程B和G拥有相同的优先级，这里讲的优先级是指获取锁的优先级，同时执行CAS指令竞争锁。如果恰好线程G成功了，线程B就得重新挂起等待被唤醒。
通过上述场景描述，我们可以看出，即使线程B等了很长时间也得和新来的线程G同时竞争锁，如此的不公平。
static final class NonfairSync extends Sync {
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
|
非公平锁加锁流程
1、第一个线程t1、第一次加锁，没有加锁之前 aqs（NonfairSync）的状态
2、t1、加锁之后
3、第二个线程t2 如果他 t1 线程没有释放 AQS状态和2一样
3.1 t2 加锁之后失败之后
3.2 t2 加锁成功之后
如果需要加锁成功则t1必须释放锁，AQS 状态回归原始
然后t2加锁成功
结论：ReentrantLock如果线程之间没有竞争，效率非常高；甚至队列都没有初始化
4、t1线程没有释放锁，t2线程在排队中，这时 t3线程来加锁
公平锁实现
在公平锁中，每当线程执行lock方法时，如果同步器的队列中有线程在等待，则直接加入到队列中。
场景分析：
- 持有锁的线程A正在running，对列中有线程BCDEF被挂起并等待被唤醒；
- 线程G执行lock，队列中有线程BCDEF在等待，线程G直接加入到队列的对尾。
所以每个线程获取锁的过程是公平的，等待时间最长的会最先被唤醒获取锁。
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
final void lock() {
acquire(1);
}
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;
}
}
|
重入锁实现
重入锁，即线程可以重复获取已经持有的锁。在非公平和公平锁中，都对重入锁进行了实现。
if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
|
条件变量Condition
条件变量很大一个程度上是为了解决Object.wait/notify/notifyAll难以使用的问题。
public class ConditionObject implements Condition, java.io.Serializable {
private transient Node firstWaiter;
private transient Node lastWaiter;
public final void signal() {}
public final void signalAll() {}
public final void awaitUninterruptibly() {}
public final void await() throws InterruptedException {}
}
|
- Synchronized中，所有的线程都在同一个object的条件队列上等待。而ReentrantLock中，每个condition都维护了一个条件队列。
- 每一个Lock可以有任意数据的Condition对象，Condition是与Lock绑定的，所以就有Lock的公平性特性：如果是公平锁，线程为按照FIFO的顺序从Condition.await中释放，如果是非公平锁，那么后续的锁竞争就不保证FIFO顺序了。
- Condition接口定义的方法，await对应于Object.wait，signal对应于Object.notify，signalAll对应于Object.notifyAll。特别说明的是Condition的接口改变名称就是为了避免与Object中的wait/notify/notifyAll的语义和使用上混淆。
先看一个condition在生产者消费者的应用场景：
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class Queue<T> {
private final T[] items;
private final Lock lock = new ReentrantLock();
private Condition notFull = lock.newCondition();
private Condition notEmpty = lock.newCondition();
private int head, tail, count;
public Queue(int maxSize) {
items = (T[]) new Object[maxSize];
}
public Queue() {
this(10);
}
public void put(T t) throws InterruptedException {
lock.lock();
try {
while (count == items.length) {
notFull.await();
}
items[tail] = t;
if (++tail == items.length) {
tail = 0;
}
++count;
notEmpty.signal();
} finally {
lock.unlock();
}
}
public T take() throws InterruptedException {
lock.lock();
try {
while (count == 0) {
notEmpty.await();
}
T o = items[head];
items[head] = null;
if (++head == items.length) {
head = 0;
}
--count;
notFull.signal();
return o;
} finally {
lock.unlock();
}
}
}
|
假设线程AB在并发的往items中插入数据，当items中元素存满时。如果线程A获取到锁，继续添加数据，满足count == items.length条件，导致线程A执行await方法。
ReentrantLock是独占锁，同一时刻只有一个线程能获取到锁，所以在lock.lock()和lock.unlock()之间可能有一次释放锁的操作（同样也必然还有一次获取锁的操作）。在Quene类中，不管take还是put，在线程持有锁之后只有await()方法有可能释放锁，然后挂起线程，一旦条件满足就被唤醒，再次获取锁。具体实现如下：
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
private Node addConditionWaiter() {
Node t = lastWaiter;
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
|
await实现逻辑：
- 将线程A加入到条件等待队列中，如果最后一个节点是取消状态，则从对列中删除。
- 线程A释放锁，实质上是线程A修改AQS的状态state为0，并唤醒AQS等待队列中的线程B，线程B被唤醒后，尝试获取锁，接下去的过程就不重复说明了。
- 线程A释放锁并唤醒线程B之后，如果线程A不在AQS的同步队列中，线程A将通过LockSupport.park进行挂起操作。
- 随后，线程A等待被唤醒，当线程A被唤醒时，会通过acquireQueued方法竞争锁，如果失败，继续挂起。如果成功，线程A从await位置恢复。
假设线程B获取锁之后，执行了take操作和条件变量的signal，signal通过某种实现唤醒了线程A，具体实现如下：
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
private void doSignal(Node first) {
do {
if ((firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
final boolean transferForSignal(Node node) {
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
|
signal实现逻辑：
- 接着上述场景，线程B执行了signal方法，取出条件队列中的第一个非CANCELLED节点线程，即线程A。另外，signalAll就是唤醒条件队列中所有非CANCELLED节点线程。遇到CANCELLED线程就需要将其从队列中删除。
- 通过CAS修改线程A的waitStatus为0，表示该节点已经不是等待条件状态，并将线程A插入到AQS的等待队列中。
- 唤醒线程A，线程A和别的线程进行锁的竞争。
总结
- ReentrantLock提供了内置锁类似的功能和内存语义。
- 此外，ReetrantLock还提供了其它功能，包括定时的锁等待、可中断的锁等待、公平性、以及实现非块结构的加锁、Condition，对线程的等待和唤醒等操作更加灵活，一个ReentrantLock可以有多个Condition实例，所以更有扩展性，不过ReetrantLock需要显示的获取锁，并在finally中释放锁，否则后果很严重。
- ReentrantLock在性能上似乎优于Synchronized，其中在jdk1.6中略有胜出，在1.5中是远远胜出。那么为什么不放弃内置锁，并在新代码中都使用ReetrantLock？
- 在java1.5中， 内置锁与ReentrantLock相比有例外一个优点：在线程转储中能给出在哪些调用帧中获得了哪些锁，并能够检测和识别发生死锁的线程。Reentrant的非块状特性任然意味着，获取锁的操作不能与特定的栈帧关联起来，而内置锁却可以。
- 因为内置锁时JVM的内置属性，所以未来更可能提升synchronized而不是ReentrantLock的性能。例如对线程封闭的锁对象消除优化，通过增加锁粒度来消除内置锁的同步。
(以上内容部分取自占小狼博客，附大佬博客地址：https://www.jianshu.com/p/4358b1466ec9)
读写锁
读写锁的基本使用
ReentrantReadWriteLock 是 ReadWriteLock的实现类。
|
|
| 公平锁和非公平锁 |
根据多线程竞争时是否排队一次获取锁，Synchronized和ReentrantLock 实现默认的都是非公平锁，非公平锁可以提高效率，避免线程唤醒带来的空档期造成CPU资源浪费。 |
| 可重入锁和不可重复锁 |
根据同一个线程是否能重复获取同一把锁。 |
| 共享锁和独占锁（排它锁） |
根据多线程是否共享一把锁，典型的就比如ReentrantLockReadWriteLock，其中读锁是共享锁，写锁是排它锁，从读锁变成写锁叫锁升级，从写锁变成读锁叫锁降级。 |
| 可中断锁和不可中断锁 |
根据正在尝试获取锁的线程是否中断。 |
| 悲观锁和乐观锁 |
根据线程是否锁住共享资源。 |
| 自旋锁和阻塞锁 |
根据线程的等待过程。 |
在ReentrantReadWriteLock中包含读锁和写锁，其中读锁是可以多线程共享的，即共享锁，而写锁是排他锁，在更改时候不允许其他线程操作。读写锁其实是一把锁，所以会有同一时刻不允许读写锁共存的规定。之所以要细分读锁和写锁也是为了提高效率，将读和写分离，对比ReentrantLock就可以发现，无论并发读还是写，它总会先锁住全部再说。
@Slf4j(topic = "ellison")
public class ReentrantLockRWTest {
private static ReentrantReadWriteLock reentrantLock = new ReentrantReadWriteLock();
private static ReentrantReadWriteLock.ReadLock readLock = reentrantLock.readLock();
private static ReentrantReadWriteLock.WriteLock writeLock = reentrantLock.writeLock();
public static void read() {
readLock.lock();
try {
log.debug(Thread.currentThread().getName() + "获取读锁，开始执行");
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
} finally {
readLock.unlock();
log.debug(Thread.currentThread().getName() + "释放读锁");
}
}
public static void write() {
writeLock.lock();
try {
log.debug(Thread.currentThread().getName() + "获取写锁，开始执行");
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
} finally {
writeLock.unlock();
log.debug(Thread.currentThread().getName() + "释放写锁");
}
}
public static void main(String[] args) {
new Thread(() -> read(), "t1").start();
new Thread(() -> read(), "t2").start();
new Thread(() -> write(), "t3").start();
new Thread(() -> write(), "t4").start();
}
}
|
输出结果如下，线程1和线程2可以同时获取读锁，而线程3和线程4只能依次获取写锁，因为线程4必须等待线程3释放写锁后才能获取到锁：
11:01:12.034 [t1] DEBUG ellison - t1获取读锁，开始执行
11:01:12.034 [t2] DEBUG ellison - t2获取读锁，开始执行
11:01:13.037 [t1] DEBUG ellison - t1释放读锁
11:01:13.037 [t3] DEBUG ellison - t3获取写锁，开始执行
11:01:13.037 [t2] DEBUG ellison - t2释放读锁
11:01:14.038 [t3] DEBUG ellison - t3释放写锁
11:01:14.038 [t4] DEBUG ellison - t4获取写锁，开始执行
11:01:15.038 [t4] DEBUG ellison - t4释放写锁
|
首先关于读写锁的并发结论是：读读并发、读写互斥、写写互斥
关于读读并发需要注意的
比如先有一个t1写锁拿到锁，后面有一些其他锁或许是读或许是写在park；
当t1释放锁之后活安装FIFO的原则唤醒等待的线程；
如果第一个被唤醒的是t2写锁则无可厚非；
不会再跟着唤醒t3，只有等t2执行完成之后才会去唤醒T3；
假设被唤醒的t3是读锁，那么t3会去判断他的下一个t4是不是读锁如果是则把t4唤醒；
t4唤醒之后会判断t5是不是读锁；如果t5也是则唤醒t5；
依次类推；
但是假设t6是写锁则不会唤醒t6了；
即使后面的t7是读锁也不会唤醒t7；
下面这个代码说明了这个现象。
@Slf4j(topic = "enjoy")
public class Lock2 {
static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
static Lock r = rwl.readLock();
static Lock w = rwl.writeLock();
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
w.lock();
try {
log.debug("t1 +");
TimeUnit.SECONDS.sleep(5);
log.debug("5s 之后");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
w.unlock();
}
}, "t1");
t1.start();
TimeUnit.SECONDS.sleep(1);
Thread t2 = new Thread(() -> {
try {
r.lock();
log.debug("t2----+锁-------");
TimeUnit.SECONDS.sleep(1);
} catch (Exception e) {
e.printStackTrace();
} finally {
log.debug("t2-----解锁-------");
r.unlock();
}
}, "t2");
t2.start();
TimeUnit.SECONDS.sleep(1);
Thread t3 = new Thread(() -> {
try {
r.lock();
log.debug("t3----+锁-------");
TimeUnit.SECONDS.sleep(1);
} catch (Exception e) {
e.printStackTrace();
} finally {
log.debug("t3----释放-------");
r.unlock();
}
}, "t3");
t3.start();
Thread t4 = new Thread(() -> {
try {
w.lock();
log.debug("t4--------+---");
TimeUnit.SECONDS.sleep(10);
log.debug("t4--------醒来---");
} catch (Exception e) {
e.printStackTrace();
} finally {
log.debug("t4--------解锁---");
w.unlock();
}
}, "t4");
t4.start();
Thread t5 = new Thread(() -> {
try {
r.lock();
log.debug("t5--------+锁---");
} catch (Exception e) {
e.printStackTrace();
} finally {
log.debug("t5--------解锁---");
r.unlock();
}
}, "t5");
t5.start();
}
}
|
读写锁之写锁上锁流程
@Slf4j(topic = "enjoy")
public class RWLock2 {
static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
static Lock r = rwl.readLock();
static Lock w = rwl.writeLock();
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
w.lock();
try {
log.debug("t1 w---加锁成功");
} finally {
w.unlock();
}
}, "t1");
t1.start();
}
}
|
下面看并发包大神Doug Lea的源码
protected final boolean tryAcquire(int acquires) {
Thread current = Thread.currentThread();
int c = getState();
int w = exclusiveCount(c);
if (c != 0) {
if (w == 0 || current != getExclusiveOwnerThread())
return false;
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
setState(c + acquires);
return true;
}
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires)){
return false;
}
setExclusiveOwnerThread(current);
return true;
}
|
读写锁之读锁上锁流程
public void lock() {
sync.acquireShared(1);
}
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
|
protected final int tryAcquireShared(int unused) {
Thread current = Thread.currentThread();
int c = getState();
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
int r = sharedCount(c);
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
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);
}
|
读锁的插队策略
设想如下场景：在非公平的ReentrantReadWriteLock锁中，线程2和线程4正在同时读取，线程3想要写入，拿不到锁（同一时刻是不允许读写锁共存的），于是进入等待队列，线程5不在队列里，现在过来想要读取，策略1是如果允许读插队，就是说线程5读先于线程3写操作执行，因为读锁是共享锁，不影响后面的线程3的写操作，这种策略可以提高一定的效率，却可能导致像线程3这样的线程一直在等待中，因为可能线程5读操作之后又来了n个线程也进行读操作，造成线程饥饿；策略2是不允许插队，即线程5的读操作必须排在线程3的写操作之后，放入队列中，排在线程3之后，这样能避免线程饥饿。
事实上ReentrantReadWriteLock在非公平情况下，读锁采用的就是策略2：不允许读锁插队，避免线程饥饿。更加确切的说是：在非公平锁情况下，允许写锁插队，也允许读锁插队，但是读锁插队的前提是队列中的头节点不能是想获取写锁的线程。
以上还在非公平ReentrantReadWriteLock锁中，在公平锁中，读写锁都是是不允许插队的，严格按照线程请求获取锁顺序执行。
下面用代码演示一下结论：在非公平锁情况下，允许写锁插队，也允许读锁插队，但是读锁插队的前提是队列中的头节点不能是想获取写锁的线程
@Slf4j(topic = "ellison")
public class ReentrantLockReadWriteTest {
private static ReentrantReadWriteLock reentrantLock = new ReentrantReadWriteLock();
private static ReentrantReadWriteLock.ReadLock readLock = reentrantLock.readLock();
private static ReentrantReadWriteLock.WriteLock writeLock = reentrantLock.writeLock();
public static void read() {
log.debug(Thread.currentThread().getName() + "开始尝试获取读锁");
readLock.lock();
try {
log.debug(Thread.currentThread().getName() + "========获取到读锁，开始执行");
Thread.sleep(20);
} catch (Exception e) {
e.printStackTrace();
} finally {
readLock.unlock();
log.debug(Thread.currentThread().getName() + "释放读锁");
}
}
public static void write() {
log.debug(Thread.currentThread().getName() + "开始尝试获取写锁");
writeLock.lock();
try {
log.debug(Thread.currentThread().getName() + "========获取到写锁，开始执行");
Thread.sleep(40);
} catch (Exception e) {
e.printStackTrace();
} finally {
log.debug(Thread.currentThread().getName() + "释放写锁");
writeLock.unlock();
}
}
public static void main(String[] args) {
new Thread(() -> write(), "t1").start();
new Thread(() -> read(), "t2").start();
new Thread(() -> read(), "t3").start();
new Thread(() -> write(), "t4").start();
new Thread(() -> read(), "t5").start();
new Thread(() -> {
Thread[] threads = new Thread[1000];
for (int i = 0; i < 1000; i++) {
threads[i] = new Thread(() -> read(), "子线程创建的Thread" + i);
}
for (int i = 0; i < 1000; i++) {
threads[i].start();
}
}).start();
}
}
|
以上测试代码就演示了，在非公平锁时，
其一：同一时刻读写锁不能同时存在。
其二：读锁非常容易插队，但前提是队列中的头结点不能是想获取写锁的线程。
总结
- 读写锁特点特点：读锁是共享锁，写锁是排他锁，读锁和写锁不能同时存在
- 插队策略：为了防止线程饥饿，读锁不能插队
- 升级策略：只能降级，不能升级
- ReentrantReadWriteLock适合于读多写少的场合，可以提高并发效率，而ReentrantLock适合普通场合
