Java 深入分析 - 并发 同步器
小概
synchronized 存在许多缺陷，使得使用非常困难，一旦开始请求锁便不能停止
- 无法中断获取过程中的线程
- 无法获取过程中的一系列状态
- 无定时功能
作为一门面向对象的语言，我们希望锁也能变得可扩展，因此如何抽象将变得十分有必要，我们应当模仿 synchronized [1] 的实现，并在细节处进行扩展
Jdk 中有一系列 同步器，几乎所有同步器都是基于 AbstractQueuedSynchronizer 实现，我们由此开始讨论
AbstractQueuedSynchronizer
AbstractQueuedSynchronizer 是一份同步模板，同步器通过封装继承该模板的匿名子类达到同步效果，该类能够省去同步器用于同步方面的很多代码，因此便将其抽象封装成了一份模板， AbstractQueuedSynchronizer 是理解同步器的重中之重
内部主要基于两方面实现
- 同步队列 - 存储排队线程
- 同步状态 - 整形，通常代表占有资源的线程数量，是如何同步的关键所在，通过同步状态的设计完成同步功能，具体含义由子类实现
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
private transient volatile Node tail;
private transient volatile Node head;
private volatile int state;
static final class Node {
volatile Node prev;
volatile Node next;
volatile Thread thread;
volatile int waitStatus;
Node nextWaiter;
// ...
}
// ...
}
同步节点等待状态分为四种
- CANCELED：因为等待超时或者中断，节点会被置为取消状态，处于取消状态的节点不会再去竞争锁，也就是说不会再被阻塞，节点会一直保持取消状态，而不会转换为其他状态。处于 CANCELED 的节点会被移出队列，被 GC 回收
- SIGNAL：表明当前的后继结点正在或者将要被阻塞，因此当前的节点被释放或者被取消时时，要唤醒它的后继结点
- CONDITION：表明当前节点在条件队列中，因为等待某个条件而被阻塞
- PROPAGATE：在共享模式下，可以认为资源有多个，因此当前线程被唤醒之后，可能还有剩余的资源可以唤醒其他线程。该状态用来表明后续节点会传播唤醒的操作。需要注意的是只有头节点才可以设置为该状态
- INITIAL：新创建的节点会处于这种状态
AbstractQueuedSynchronizer 支持 独占式 或 共享式 访问资源，但 如何获取和释放并没有实现，这和同步状态息息相关，默认交给子类
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}
protected boolean tryReleaseShared(int arg) {
throw new UnsupportedOperationException();
}
独占式获取与释放
独占式意味着只有一个线程可以占用资源，并排斥其他线程
AbstractQueuedSynchronizer 中独占式获取，仅有 head 占有资源，其后继节点将被挂起
独占式释放后，会搜索第一个非 CANCELLED 节点，并将其唤醒
共享式获取与释放
独占式访问资源时，排斥其他访问
共享式访问资源时，其他共享式的访问均被允许，而独占式应当被排斥
中断获取
AbstractQueuedSynchronizer 支持获取时对中断的响应，在获取途中如若发生中断，会立即 抛出中断异常，那么获取操作将马上得到释放
超时获取
指定超时时间，如果 获取超过限定时间，获取操作会马上返回
我们如此次假设，超时时间为 timeout，执行一次获取操作前的时间为 begin，此次获取失败后的时间为 now
那么还能够执行获取操作的时间，就应当限定在 timeout -= (now - begin) 内，超出时间范围需立马返回
Condition
wait / notify / notifyAll 与 synchronized 配套使用
那么这里相对应的就是 await / signal / signalAll，在 Condition 接口中，还支持 非中断响应等待，超时等待
关于 Condition 的实现部分和 synchronized 其实是差不多的
能够等待或唤醒的线程，必定是已经获取锁的线程，此处不需要额外的 CAS 操作
Condition 内部维护了一个 等待队列
- 等待时，创建当前线程节点并 enqueue 等待队列，释放同步状态，唤醒同步队列后继节点，当前线程将被挂起，如果 碰到中断将直接抛出异常
- 唤醒时，从等待队列 poll 一个节点，enqueue 同步队列
- 唤醒全部时，对等待队列全部节点执行唤醒操作
public class ConditionObject implements Condition, java.io.Serializable {
private static final long serialVersionUID = 1173984872572414699L;
/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter;
// ...
}
如何实现获取与释放
实现获取与释放，应当基于 控制同步状态，我们通过操作同步状态来达到需要的特定要求，下面给出一个例子，例中为一把最大两个线程同时占用的共享锁，其中同步状态表示占有资源的线程个数，只有同步状态在限定范围内时才会返回，由此来完成共享逻辑
public class TwinsLock implements Lock {
private final SharedSyn syn = new SharedSyn(2);
@Override
public void lock() {
syn.acquireShared(1);
}
@Override
public void unlock() {
syn.releaseShared(1);
}
// ...
private static class SharedSyn extends AbstractQueuedSynchronizer {
private final int maxCount;
Syn(int maxCount) {
this.maxCount = maxCount;
}
@Override
public int tryAcquireShared(int count) {
while (true) {
int currentCount = getState();
int newCount = currentCount + count;
if (newCount <= maxCount &&
compareAndSetState(currentCount, newCount)) {
return newCount;
}
}
}
@Override
public boolean tryReleaseShared(int count) {
while (true) {
int currentCount = getState();
int newCount = currentCount - count;
if (compareAndSetState(currentCount, newCount)) {
return true;
}
}
}
}
}
ReentrantLock
ReentrantLock 具有以下性质
- 独占锁
- 可重入
- 可轮询获取
- 可中断获取
- 可超时获取
- 可公平或者非公平获取
内部基于 AbstractQueuedSynchronizer 实现，在 ReentrantLock 中，同步状态表示占有资源的线程个数
/**
* Base of synchronization control for this lock. Subclassed
* into fair and nonfair versions below. Uses AQS state to
* represent the number of holds on the lock.
*/
abstract static class Sync extends AbstractQueuedSynchronizer {...}
/**
* Sync object for non-fair locks
*/
static final class NonfairSync extends Sync {...}
/**
* Sync object for fair locks
*/
static final class FairSync extends Sync {...}
可重入
当一个持有锁的线程，再次访问资源时应当允许访问，即为可重入
ReentrantLock 中，当前 state != 0 时，即有线程在占用资源，然后发现 持有锁的人就是自己，将会允许入内，并且 state++
在释放时，资源的 最终释放将会在 state = 0 时
公平与非公平
ReentrantLock 实现了两种获取机制，一种公平获取，另一种非公平获取，默认为非公平竞争模式
公平获取相当与排队买饭 - 进入等待同步队列
非公平获取相当与插队抢饭 - 绕过同步队列，直接尝试获取，失败再进入等待队列
非公平锁在被唤醒时也可能直接插队
ReentrantReadWriteLock
前面提到的 ReentrantLock 和 synchronized 都是独占锁，而此时的 ReentrantReadWriteLock 是独占式和共享式共存的一种锁，执行写时是独占模式，执行读时是共享模式
- 独占锁
- 共享锁
- 可重入
- 可轮询获取
- 可中断获取
- 可超时获取
- 可公平或者非公平获取
也就是说，写时排斥一切操作，读时容纳其他读操作，但排斥写操作，因此，在读多写少的情况下，ReentrantReadWriteLock 同步器效率将会非常高
ReentrantReadWriteLock 由于存在两种模式，如何设计读写状态变得十分关键
读写状态
在 ReentrantLock 中，只存在一种模式，因此同步状态可表示为占有资源的线程数量，但 ReentrantReadWriteLock 中却两种模式并存
最简单的一种方式，就是将身为 Integer 的 同步状态分成两部分，高 16 位代表共享式，低 16 位代表独占式
设同步状态为 state
- 共享式：state & 0xffff0000
- 独占式：state & 0x0000ffff
获取与释放
我们通过一个具体的例子来讨论，ReentrantReadWriteLock 应该如何实现获取与释放
读操作用 R 表示，写操作用 W 表示，此时同步队列如下表所示，如果全是 不同线程的访问，那么：W -> 3R -> W -> 2R
读写锁在获取过程中也是可重入的，也就是说持有锁的同一线程，是能重复获取锁的
锁降级
以上例子代表读写分开的操作，但 如果一个操作将读与写杂糅在一起，如果要保证这一操作的原子性，比如写完后马上读，必须在写锁释放前获取到读锁，不然其他线程很可能在这一瞬间占领写锁，原子性便会从此丧失，这种操作叫做 锁降级
public void writeAndRead() {
writeLock.lock();
try {
// write...
readLock.lock();
} finally {
writeLock.unlock();
}
try {
// read...
} finally {
readLock.unlock();
}
}
其他同步器
除了上面描述的两种同步器，Jdk 中还有许多同步器，如 CyclicBarrier，CountDownLatch， Semaphore ... 我们在这里就不一一阐述了，有兴趣可以看看源码
参考
- 《Java并发编程的艺术》
