高并发之Phaser、ReadWriteLock、StampedLock
本系列研究总结高并发下的几种同步锁的使用以及之间的区别,分别是:ReentrantLock、CountDownLatch、CyclicBarrier、Phaser、ReadWriteLock、StampedLock、Semaphore、Exchanger、LockSupport。由于博客园对博客字数的要求限制,会分为三个篇幅:
高并发之ReentrantLock、CountDownLatch、CyclicBarrier
高并发之Phaser、ReadWriteLock、StampedLock
高并发之Semaphore、Exchanger、LockSupport
Phaser
Phaser是JDK7开始引入的一个同步工具类，适用于一些需要分阶段的任务的处理。它的功能与 CyclicBarrier和CountDownLatch有些类似，功能上与 CountDownLatch 和 CyclicBarrier类似但支持的场景更加灵活类似于一个多阶段的栅栏，并且功能更强大，我们来比较下这三者的功能：
| 同步器 | 作用 |
|---|---|
| CountDownLatch | 倒数计数器，初始时设定计数器值，线程可以在计数器上等待，当计数器值归0后，所有等待的线程继续执行 |
| CyclicBarrier | 循环栅栏，初始时设定参与线程数，当线程到达栅栏后，会等待其它线程的到达，当到达栅栏的总数满足指定数后，所有等待的线程继续执行 |
| Phaser | 多阶段栅栏，可以在初始时设定参与线程数，也可以中途注册/注销参与者，当到达的参与者数量满足栅栏设定的数量后，会进行阶段升级（advance） |
使用场景
相对于前面的CyclicBarrier和CountDownLatch而言,这个稍微有一些难以理解,这儿引入一个场景:结婚
一场婚礼中势必分成很多个阶段,例如宾客到齐、举行婚礼、新郎新娘拜天地、入洞房、吃宴席、宾客离开等,如果把不同的人看成是不同的线程的话,那么不同的线程所要到的阶段是不一样的,例如新郎新娘可能要走完全流程,而宾客可能只是其中的几步而已。
代码示例:
Person
static class Person {
String name;
public Person(String name) {
this.name = name;
}
public void arrive() {
milliSleep(r.nextInt(1000));
System.out.printf("%s 到达现场！\n", name);
}
public void eat() {
milliSleep(r.nextInt(1000));
System.out.printf("%s 吃完!\n", name);
}
public void leave() {
milliSleep(r.nextInt(1000));
System.out.printf("%s 离开！\n", name);
}
}
}
MarriagePhaser
static class MarriagePhaser extends Phaser {
@Override
protected boolean onAdvance(int phase, int registeredParties) {
switch (phase) {
case 0:
System.out.println("所有人到齐了！");
return false;
case 1:
System.out.println("所有人吃完了！");
return false;
case 2:
System.out.println("所有人离开了！");
System.out.println("婚礼结束！");
return true;
default:
return true;
}
}
}
TestPhaser
public class TestPhaser {
static Random r = new Random();
static MarriagePhaser phaser = new MarriagePhaser();
static void milliSleep(int milli) {
try {
TimeUnit.MILLISECONDS.sleep(milli);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
phaser.bulkRegister(5);
for(int i=0; i<5; i++) {
final int nameIndex = i;
new Thread(()->{
Person p = new Person("person " + nameIndex);
p.arrive();
phaser.arriveAndAwaitAdvance();
p.eat();
phaser.arriveAndAwaitAdvance();
p.leave();
phaser.arriveAndAwaitAdvance();
}).start();
}
}
打印结果
person 0 到达现场！
person 2 到达现场！
person 4 到达现场！
person 1 到达现场！
person 3 到达现场！
所有人到齐了！
person 2 吃完!
person 0 吃完!
person 4 吃完!
person 3 吃完!
person 1 吃完!
所有人吃完了！
person 3 离开！
person 1 离开！
person 0 离开！
person 4 离开！
person 2 离开！
所有人离开了！
婚礼结束！
Phaser常见的方法
Phaser() //默认的构造方法，初始化注册的线程数量为0
Phaser(int parties)//一个指定线程数量的构造方法
此外Phaser还支持Tiering类型具有父子关系的构造方法，主要是为了减少在注册者数量庞大的时候，通过分组的形式复用Phaser从而减少竞争，提高吞吐，这种形式一般不常见，所以这里不再提及，有兴趣的可以参考官网文档。
其他几个常见方法：
register()//添加一个新的注册者
bulkRegister(int parties)//添加指定数量的多个注册者
arrive()// 到达栅栏点直接执行，无须等待其他的线程
arriveAndAwaitAdvance()//到达栅栏点，必须等待其他所有注册者到达
arriveAndDeregister()//到达栅栏点，注销自己无须等待其他的注册者到达
onAdvance(int phase, int registeredParties)//多个线程达到注册点之后，会调用该方法。
- arriveAndAwaitAdvance() 当前线程当前阶段执行完毕，等待其它线程完成当前阶段。如果当前线程是该阶段最后一个未到达的，则该方法直接返回下一个阶段的序号（阶段序号从0开始），同时其它线程的该方法也返回下一个阶段的序号。
- arriveAndDeregister() 该方法立即返回下一阶段的序号，并且其它线程需要等待的个数减一，并且把当前线程从之后需要等待的成员中移除。如果该Phaser是另外一个Phaser的子Phaser（层次化Phaser会在后文中讲到），并且该操作导致当前Phaser的成员数为0，则该操作也会将当前Phaser从其父Phaser中移除。
- arrive()该方法不作任何等待，直接返回下一阶段的序号。
- awaitAdvance(int phase) 该方法等待某一阶段执行完毕。如果当前阶段不等于指定的阶段或者该Phaser已经被终止，则立即返回。该阶段数一般由arrive()方法或者arriveAndDeregister()方法返回。返回下一阶段的序号，或者返回参数指定的值（如果该参数为负数），或者直接返回当前阶段序号（如果当前Phaser已经被终止）。
- awaitAdvanceInterruptibly(int phase) 效果与awaitAdvance(int phase)相当，唯一的不同在于若该线程在该方法等待时被中断，则该方法抛出InterruptedException。
- awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit) 效果与awaitAdvanceInterruptibly(int phase)相当，区别在于如果超时则抛出TimeoutException。
- bulkRegister(int parties) 注册多个party。如果当前phaser已经被终止，则该方法无效，并返回负数。如果调用该方法时，onAdvance方法正在执行，则该方法等待其执行完毕。如果该Phaser有父Phaser则指定的party数大于0，且之前该Phaser的party数为0，那么该Phaser会被注册到其父Phaser中。
- forceTermination() 强制让该Phaser进入终止状态。已经注册的party数不受影响。如果该Phaser有子Phaser，则其所有的子Phaser均进入终止状态。如果该Phaser已经处于终止状态，该方法调用不造成任何影响。
ReadWriteLock
根据翻译,读写锁,顾名思义，在读的时候上读锁，在写的时候上写锁，这样就很巧妙的解决synchronized的一个性能问题：读与读之间互斥。
ReadWriteLock也是一个接口，原型如下：
public interface ReadWriteLock {
Lock readLock();
Lock writeLock();
}
该接口只有两个方法，读锁和写锁。也就是说，我们在写文件的时候，可以将读和写分开，分成2个锁来分配给线程，从而可以做到读和读互不影响，读和写互斥，写和写互斥，提高读写文件的效率。该接口也有一个实现类ReentrantReadWriteLock，下面我们就来学习下这个类。
我们先看一下，多线程同时读取文件时，用synchronized实现的效果，代码如下：
public class ReadAndWriteLock {
public synchronized void get(Thread thread) {
long start = System.currentTimeMillis();
for(int i=0; i<5; i++){
try {
Thread.sleep(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(thread.getName() + ":正在进行读操作……");
}
System.out.println(thread.getName() + ":读操作完毕！");
long end = System.currentTimeMillis();
System.out.println("用时:"+(end-start)+"ms");
}
public static void main(String[] args) {
final ReadAndWriteLock lock = new ReadAndWriteLock();
new Thread(new Runnable() {
@Override
public void run() {
lock.get(Thread.currentThread());
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
lock.get(Thread.currentThread());
}
}).start();
}
}
测试结果如下：
Thread-1:正在进行读操作……
Thread-1:正在进行读操作……
Thread-1:正在进行读操作……
Thread-1:正在进行读操作……
Thread-1:正在进行读操作……
Thread-1:读操作完毕！
用时:112ms
Thread-0:正在进行读操作……
Thread-0:正在进行读操作……
Thread-0:正在进行读操作……
Thread-0:正在进行读操作……
Thread-0:正在进行读操作……
Thread-0:读操作完毕！
用时:107ms
我们可以看到，即使是在读取文件，在加了synchronized关键字之后，读与读之间，也是互斥的，也就是说，必须等待Thread-0读完之后，才会轮到Thread-1线程读，而无法做到同时读文件，这种情况在大量线程同时都需要读文件的时候，读写锁的效率，明显要高于synchronized关键字的实现。下面我们来测试一下，代码如下：
public class ReadAndWriteLock {
ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
public void get(Thread thread) {
lock.readLock().lock();
try{
System.out.println("start time:"+System.currentTimeMillis());
for(int i=0; i<5; i++){
try {
Thread.sleep(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(thread.getName() + ":正在进行读操作……");
}
System.out.println(thread.getName() + ":读操作完毕！");
System.out.println("end time:"+System.currentTimeMillis());
}finally{
lock.readLock().unlock();
}
}
public static void main(String[] args) {
final ReadAndWriteLock lock = new ReadAndWriteLock();
new Thread(new Runnable() {
@Override
public void run() {
lock.get(Thread.currentThread());
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
lock.get(Thread.currentThread());
}
}).start();
}
}
注意的是，如果有一个线程已经占用了读锁，则此时其他线程如果要申请写锁，则申请写锁的线程会一直等待释放读锁。如果有一个线程已经占用了写锁，则此时其他线程如果申请写锁或者读锁，则申请的线程会一直等待释放写锁。读锁和写锁是互斥的。
下面我们来验证下读写锁的互斥关系，代码如下：
public class ReadAndWriteLock {
ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
public static void main(String[] args) {
final ReadAndWriteLock lock = new ReadAndWriteLock();
// 建N个线程，同时读
ExecutorService service = Executors.newCachedThreadPool();
service.execute(new Runnable() {
@Override
public void run() {
lock.readFile(Thread.currentThread());
}
});
// 建N个线程，同时写
ExecutorService service1 = Executors.newCachedThreadPool();
service1.execute(new Runnable() {
@Override
public void run() {
lock.writeFile(Thread.currentThread());
}
});
}
// 读操作
public void readFile(Thread thread){
lock.readLock().lock();
boolean readLock = lock.isWriteLocked();
if(!readLock){
System.out.println("当前为读锁！");
}
try{
for(int i=0; i<5; i++){
try {
Thread.sleep(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(thread.getName() + ":正在进行读操作……");
}
System.out.println(thread.getName() + ":读操作完毕！");
}finally{
System.out.println("释放读锁！");
lock.readLock().unlock();
}
}
// 写操作
public void writeFile(Thread thread){
lock.writeLock().lock();
boolean writeLock = lock.isWriteLocked();
if(writeLock){
System.out.println("当前为写锁！");
}
try{
for(int i=0; i<5; i++){
try {
Thread.sleep(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(thread.getName() + ":正在进行写操作……");
}
System.out.println(thread.getName() + ":写操作完毕！");
}finally{
System.out.println("释放写锁！");
lock.writeLock().unlock();
}
}
}
测试结果如下：
// 读锁和读锁测试结果：
当前为读锁！
当前为读锁！
pool-2-thread-1:正在进行读操作……
pool-1-thread-1:正在进行读操作……
pool-2-thread-1:正在进行读操作……
pool-1-thread-1:正在进行读操作……
pool-2-thread-1:正在进行读操作……
pool-1-thread-1:正在进行读操作……
pool-2-thread-1:正在进行读操作……
pool-1-thread-1:正在进行读操作……
pool-1-thread-1:正在进行读操作……
pool-2-thread-1:正在进行读操作……
pool-1-thread-1:读操作完毕！
pool-2-thread-1:读操作完毕！
释放读锁！
释放读锁！
// 测试结果不互斥
// 读锁和写锁，测试结果如下：
当前为读锁！
pool-1-thread-1:正在进行读操作……
pool-1-thread-1:正在进行读操作……
pool-1-thread-1:正在进行读操作……
pool-1-thread-1:正在进行读操作……
pool-1-thread-1:正在进行读操作……
pool-1-thread-1:读操作完毕！
释放读锁！
当前为写锁！
pool-2-thread-1:正在进行写操作……
pool-2-thread-1:正在进行写操作……
pool-2-thread-1:正在进行写操作……
pool-2-thread-1:正在进行写操作……
pool-2-thread-1:正在进行写操作……
pool-2-thread-1:写操作完毕！
释放写锁！
// 测试结果互斥
// 写锁和写锁，测试结果如下：
当前为写锁！
pool-1-thread-1:正在进行写操作……
pool-1-thread-1:正在进行写操作……
pool-1-thread-1:正在进行写操作……
pool-1-thread-1:正在进行写操作……
pool-1-thread-1:正在进行写操作……
pool-1-thread-1:写操作完毕！
释放写锁！
当前为写锁！
pool-2-thread-1:正在进行写操作……
pool-2-thread-1:正在进行写操作……
pool-2-thread-1:正在进行写操作……
pool-2-thread-1:正在进行写操作……
pool-2-thread-1:正在进行写操作……
pool-2-thread-1:写操作完毕！
释放写锁！
// 测试结果互斥
ReadWriteLock小结
使用ReadWriteLock可以提高读取效率：
-
ReadWriteLock只允许一个线程写入； -
ReadWriteLock允许多个线程在没有写入时同时读取； -
ReadWriteLock适合读多写少的场景。
StampedLock
前面介绍的ReadWriteLock可以解决多线程同时读，但只有一个线程能写的问题。
如果我们深入分析ReadWriteLock，会发现它有个潜在的问题：如果有线程正在读，写线程需要等待读线程释放锁后才能获取写锁，即读的过程中不允许写，这是一种悲观的读锁。
要进一步提升并发执行效率，Java 8引入了新的读写锁：StampedLock。
StampedLock和ReadWriteLock相比，改进之处在于：读的过程中也允许获取写锁后写入！这样一来，我们读的数据就可能不一致，所以，需要一点额外的代码来判断读的过程中是否有写入，这种读锁是一种乐观锁。
乐观锁的意思就是乐观地估计读的过程中大概率不会有写入，因此被称为乐观锁。反过来，悲观锁则是读的过程中拒绝有写入，也就是写入必须等待。显然乐观锁的并发效率更高，但一旦有小概率的写入导致读取的数据不一致，需要能检测出来，再读一遍就行。
我们来看例子：
public class Point {
private final StampedLock stampedLock = new StampedLock();
private double x;
private double y;
public void move(double deltaX, double deltaY) {
long stamp = stampedLock.writeLock(); // 获取写锁
try {
x += deltaX;
y += deltaY;
} finally {
stampedLock.unlockWrite(stamp); // 释放写锁
}
}
public double distanceFromOrigin() {
long stamp = stampedLock.tryOptimisticRead(); // 获得一个乐观读锁
// 注意下面两行代码不是原子操作
// 假设x,y = (100,200)
double currentX = x;
// 此处已读取到x=100，但x,y可能被写线程修改为(300,400)
double currentY = y;
// 此处已读取到y，如果没有写入，读取是正确的(100,200)
// 如果有写入，读取是错误的(100,400)
if (!stampedLock.validate(stamp)) { // 检查乐观读锁后是否有其他写锁发生
stamp = stampedLock.readLock(); // 获取一个悲观读锁
try {
currentX = x;
currentY = y;
} finally {
stampedLock.unlockRead(stamp); // 释放悲观读锁
}
}
return Math.sqrt(currentX * currentX + currentY * currentY);
}
}
和ReadWriteLock相比，写入的加锁是完全一样的，不同的是读取。注意到首先我们通过tryOptimisticRead()获取一个乐观读锁，并返回版本号。接着进行读取，读取完成后，我们通过validate()去验证版本号，如果在读取过程中没有写入，版本号不变，验证成功，我们就可以放心地继续后续操作。如果在读取过程中有写入，版本号会发生变化，验证将失败。在失败的时候，我们再通过获取悲观读锁再次读取。由于写入的概率不高，程序在绝大部分情况下可以通过乐观读锁获取数据，极少数情况下使用悲观读锁获取数据。
可见，StampedLock把读锁细分为乐观读和悲观读，能进一步提升并发效率。但这也是有代价的：
一是代码更加复杂
二是StampedLock是不可重入锁，不能在一个线程中反复获取同一个锁。
StampedLock还提供了更复杂的将悲观读锁升级为写锁的功能，它主要使用在if-then-update的场景：即先读，如果读的数据满足条件，就返回，如果读的数据不满足条件，再尝试写。
StampedLock小结
StampedLock提供了乐观读锁，可取代ReadWriteLock以进一步提升并发性能；
StampedLock是不可重入锁。