Java高并发之锁的使用以及原理浅析
Java高并发之锁的使用以及原理浅析
锁像synchronized同步块一样，是一种线程同步机制。让自Java 5开始，java.util.concurrent.locks包提供了另一种方式实现线程同步机制——Lock。那么问题来了既然都可以通过synchronized来实现同步访问了，那么为什么还需要提供Lock呢？这个问题我们下面讨论java.util.concurrent.locks包中包含了一些锁的实现，所以我们不需要重复造轮子了。但是我们仍然需要去了解怎样使用这些锁，且了解这些实现背后的理论也是很有用处的。
本文将从下面几个方面介绍
- 锁的相关概念
- java.util.concurrent.locks下常用的几种锁
锁的相关概念
在学习或者使用Java的过程中进程会遇到各种各样的锁的概念：公平锁、非公平锁、自旋锁、可重入锁、偏向锁、轻量级锁、重量级锁、读写锁、互斥锁等待。下边总结了对各种锁的解释
公平锁/非公平锁
公平锁是指多个线程在等待同一个锁时按照申请锁的先后顺序来获取锁。相反的非公平锁是指多个线程获取锁的顺序并不是按照申请锁的顺序，有可能后申请的线程比先申请的线程优先获取锁。
公平锁的好处是等待锁的线程不会饿死，但是整体效率相对低一些；非公平锁的好处是整体效率相对高一些，但是有些线程可能会饿死或者说很早就在等待锁，但要等很久才会获得锁。其中的原因是公平锁是严格按照请求所的顺序来排队获得锁的，而非公平锁时可以抢占的，即如果在某个时刻有线程需要获取锁，而这个时候刚好锁可用，那么这个线程会直接抢占，而这时阻塞在等待队列的线程则不会被唤醒。
对于Java ReentrantLock而言，通过构造函数指定该锁是否是公平锁，默认是非公平锁。例：new ReentrantLock(true)是公平锁
对于Synchronized而言，也是一种非公平锁。由于其并不像ReentrantLock是通过AQS的来实现线程调度，所以并没有任何办法使其变成公平锁。
可重入锁
也叫递归锁，是指在外层函数获得锁之后，内层递归函数仍然可以获取到该锁。即线程可以进入任何一个它已经拥有锁的代码块。在JAVA环境下 ReentrantLock 和synchronized 都是可重入锁。可重入锁最大的作用是避免死锁。
具体区别下文阐述。
自旋锁
在Java中，自旋锁是指尝试获取锁的线程不会立即阻塞，而是采用循环的方式去尝试获取锁，这样的好处是减少线程上下文切换的消耗，缺点是循环会消耗CPU。
JDK6中已经变为默认开启自旋锁，并且引入了自适应的自旋锁。自适应意味着自旋的时间不在固定了，而是由前一次在同一个锁上的自旋时间及锁的拥有者的状态来决定。自旋是在轻量级锁中使用的，在重量级锁中，线程不使用自旋。
偏向锁、轻量级锁和重量级锁
这三种锁是指锁的状态，并且是针对Synchronized。在Java 5后通过引入锁升级的机制来实现高效Synchronized。这三种锁的状态是通过对象监视器在对象头中的字段来表明的。如下图
这里的无锁和偏向锁在对象头的倒数第三bit中分别采用0和1标记
- 偏向锁是JDK6中引入的一项锁优化，它的目的是消除数据在无竞争情况下的同步原语，进一步提高程序的运行性能。偏向锁会偏向于第一个获得它的线程，如果在接下来的执行过程中，该锁没有被其他的线程获取，则持有偏向锁的线程将永远不需要同步。但是对于锁竞争激励的场合，我其效果不佳。最坏的情况下就是每次都是不同的线程来请求相同的锁，这样偏向模式就会失效。
- 轻量级锁是指当锁是偏向锁的时候，被另一个线程所访问，偏向锁就会升级为轻量级锁，其他线程会通过自旋的形式尝试获取锁，不会阻塞，提高性能。
- 重量级锁是指当锁为轻量级锁的时候，另一个线程虽然是自旋，但自旋不会一直持续下去，当自旋一定次数的时候，还没有获取到锁，就会进入阻塞，该锁膨胀为重量级锁。重量级锁会让其他申请的线程进入阻塞，性能降低。
悲观锁和乐观锁
乐观锁与悲观锁不是指具体的什么类型的锁，而是指看待并发同步的角度
- 乐观锁认为对于同一个数据的并发操作，是不会发生修改的。在更新数据的时候，会采用尝试更新，不断重新的方式更新数据。乐观的认为，不加锁的并发操作是没有事情的。即假定不会发生并发冲突，只在提交操作时检测是否违反数据完整性。（使用版本号或者时间戳来配合实现）。在java中就是 是无锁编程，常常采用的是CAS算法，典型的例子就是原子类，通过CAS自旋实现原子操作的更新。
- 悲观锁认为对于同一个数据的并发操作，一定是会发生修改的，哪怕没有修改，也会认为修改。因此对于同一个数据的并发操作，悲观锁采取加锁的形式。悲观的认为，不加锁的并发操作一定会出问题。即假定会发生并发冲突，屏蔽一切可能违反数据完整性的操作。在java中就是各种锁编程。
- 从上面的描述我们可以看出，悲观锁适合写操作非常多的场景，乐观锁适合读操作非常多的场景，不加锁会带来大量的性能提升。
共享锁和独占锁
- 共享锁：如果事务T对数据A加上共享锁后，则其他事务只能对A再加共享锁，不能加排它锁。获准共享锁的事务只能读数据，不能修改数据。
- 独占锁：如果事务T对数据A加上独占锁后，则其他事务不能再对A加任何类型的锁。获得独占锁的事务即能读数据又能修改数据。如Synchronized
互斥锁和读写锁
独占锁/共享锁就是一种广义的说法，互斥锁/读写锁就是具体的实现。
- 互斥锁：就是指一次最多只能有一个线程持有的锁。在JDK中synchronized和JUC的Lock就是互斥锁。
- 读写锁：读写锁是一个资源能够被多个读线程访问，或者被一个写线程访问但不能同时存在读线程。Java当中的读写锁通过ReentrantReadWriteLock实现。ReentrantReadWriteLock运行一个资源可以被多个读操作访问，或者一个写操作访问，但两者不能同时进行。
java.util.concurrent.locks下常用的几种锁
ReentrantLock
ReentrantLock，可重入锁，是一种递归无阻塞的同步机制。它可以等同于synchronized的使用，但是ReentrantLock提供了比synchronized更强大、灵活的锁机制，可以减少死锁发生的概率。
ReentrantLock还提供了公平锁和非公平锁的选择，构造方法接受一个可选的公平参数（默认非公平锁），当设置为true时，表示公平锁，否则为非公平锁。
获取锁
一般使用如下方式获取锁
ReentrantLock lock = new ReentrantLock(); lock.lock();
lock方法：
public void lock() { sync.lock(); }
Sync为Sync为ReentrantLock里面的一个内部类，它继承AQS。关于AQS的相关知识可以自行补充一下。Sync有两个子类分别是FairSync（公平锁）和 NofairSync（非公平锁）。默认使用NofairSync，下面是ReentrantLock的构造类
public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); }
下边是一个简单的重入锁使用案例
public class ReentrantLockDemo implements Runnable {
public static final Lock lock = new ReentrantLock();
public static int i = 0;
@Override
public void run() {
for (int j = 0; j < 1000000; j++) {
lock.lock();
try {
i++;
} finally {
lock.unlock();
}
}
}
public static void main(String[] args) throws InterruptedException {
ReentrantLockDemo demo = new ReentrantLockDemo();
Thread t1 = new Thread(demo);
Thread t2 = new Thread(demo);
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println(i);
}
}
上述代码的第8~12行,使用了重入锁保护了临界区资源i，确保了多线程对i的操作。输出结果为2000000。可以看到与synchronized相比，重入锁必选手动指定在什么地方加锁，什么地方释放锁，所以更加灵活。
要注意是，再退出临界区的时候，需要释放锁，否则其他线程就无法访问临界区了。这里为啥叫可重入锁是因为这种锁是可以被同一个线程反复进入的。比如上述代码��使用锁部分可以写成这样
lock.lock(); lock.lock(); try { i++; } finally { lock.unlock(); lock.unlock(); }
在这种情况下，一个线程联连续两次获取同一把锁，这是允许的。但是需要注意的是，如果同一个线程多次获的锁，那么在释放是也要释放相同次数的锁。如果释放的锁少了，相当于该线程依然持有这个锁，那么其他线程就无法访问临界区了。释放的次数多了也会抛出java.lang.IllegalMonitorStateException异常。
除了使用上的灵活，ReentrantLock还提供了一些高级功能如中断。限时等待等。
中断响应
对用synchrozide来说，如果一个线程在等待，那么结果只有两种情况，要么获得这把锁继续执行下去要么一直等待下去。而使用重入锁，提供了另外一种可能，那就是线程可以被中断。也就是说在这里可以取消对锁的请求。这种情况对解决死锁是有一定帮组的。
下面代码产生了一个死锁，但是我们可以通过锁的中断，解决这个死锁。
public class ReentrantLockDemo implements Runnable { //重入锁ReentrantLock public static ReentrantLock lock1 = new ReentrantLock(); public static ReentrantLock lock2 = new ReentrantLock(); int lock; public ReentrantLockDemo(int lock) { this.lock = lock; } @Override public void run() { try { if (lock == 1) { lock1.lockInterruptibly(); Thread.sleep(500); lock2.lockInterruptibly(); System.out.println("this is thread 1"); } else { lock2.lockInterruptibly(); Thread.sleep(500); lock1.lockInterruptibly(); System.out.println("this is thread 2"); } } catch (Exception e) { //e.printStackTrace(); } finally { if (lock1.isHeldByCurrentThread()) { lock1.unlock();//释放锁 } if (lock2.isHeldByCurrentThread()) { lock2.unlock(); } System.out.println(Thread.currentThread().getId() + "：线程退出"); } } public static void main(String[] args) throws InterruptedException { ReentrantLockDemo r1 = new ReentrantLockDemo(1); ReentrantLockDemo r2 = new ReentrantLockDemo(2); Thread t1 = new Thread(r1); Thread t2 = new Thread(r2); t1.start(); t2.start(); Thread.sleep(1000); //t2线程被中断，放弃锁申请，释放已获得的lock2，这个操作使得t1线程顺利获得lock2继续执行下去； //若没有此段代码，t2线程没有中断，那么会出现t1获取lock1，请求lock2，而t2获取lock2，请求lock1的相互等待死锁情况 t2.interrupt(); } }
线程t1和t2启动后，t1先占用lock1然后在请求lock2；t2先占用lock2，然后请求lock1，因此很容易形成线程之间的相互等待。着这里使用的是ReenTrantLock提供了一种能够中断等待锁的线程的机制，通过lock.lockInterruptibly()来实现这个机制。
最后由于t2线程被中断，t2会放弃对lock1的1请求，同时释放lock2。这样可以使t1继续执行下去，结果如下图
锁申请等待限时
除了等待通知以外，避免死锁还有另外一种方式，那就是限时等待。通过给定一个等待时间，让线程自动放弃。
public class TimeLockDemo implements Runnable { private static ReentrantLock reentrantLock = new ReentrantLock(); @Override public void run() { try { if (reentrantLock.tryLock(5, TimeUnit.SECONDS)) { Thread.sleep(6000); } else { System.out.println("Gets lock failed"); } } catch (InterruptedException e) { e.printStackTrace(); } finally { if (reentrantLock.isHeldByCurrentThread()){ reentrantLock.unlock(); } } } public static void main(String[] args) { TimeLockDemo demo1 = new TimeLockDemo(); TimeLockDemo demo2 = new TimeLockDemo(); Thread t1 = new Thread(demo1); Thread t2 = new Thread(demo2); t1.start(); t2.start(); } }
tryLock有两个参数，一个表示等待时长，另一个表示计时单位。在这里就是通过lock.tryLock(5,TimeUnit.SECONDS)来设置锁申请等待限时，此例就是限时等待5秒获取锁。在这里的锁请求最多为5秒，如果超过5秒未获得锁请求，则会返回fasle，如果成功获得锁就会返回true。此案例中第一个线程会持有锁长达6秒，所以另外一个线程无法在5秒内获得锁 故案例输出结果为Gets lock failed
另外tryLock方法也可以不带参数之直接运行，在这种情况下，当前线程会尝试获得锁，如果锁并未被其他线程占用，则申请锁直接成功，立即返回true，否则当前线程不会进行等待，而是立即返回false。这种模式不会引起线程等待，因此也不会产生死锁。
下边展示了这种使用方式
public class ReentrantLockDemo implements Runnable { //重入锁ReentrantLock public static ReentrantLock lock1 = new ReentrantLock(); public static ReentrantLock lock2 = new ReentrantLock(); int lock; public ReentrantLockDemo(int lock) { this.lock = lock; } @Override public void run() { try { if (lock == 1) { while (true) { if (lock1.tryLock()) { try { Thread.sleep(1000); } finally { lock1.unlock(); } } if (lock2.tryLock()) { try { System.out.println("thread " + Thread.currentThread().getId() + " 执行完毕"); return; } finally { lock2.unlock(); } } } } else { while (true) { if (lock2.tryLock()) { try { Thread.sleep(1000); } finally { lock2.unlock(); } } if (lock1.tryLock()) { try { System.out.println("thread " + Thread.currentThread().getId() + " 执行完毕"); return; } finally { lock1.unlock(); } } } } } catch (Exception e) { e.printStackTrace(); } } public static void main(String[] args) throws InterruptedException { ReentrantLockDemo r1 = new ReentrantLockDemo(1); ReentrantLockDemo r2 = new ReentrantLockDemo(2); Thread t1 = new Thread(r1); Thread t2 = new Thread(r2); t1.start(); t2.start(); } }View Code
使用了tryLock后，线程不会傻傻的等待，而是不同的尝试获取锁，因此，只要执行足够长的时间，线程总是会获得所有需要的资源。从而正常执行。下边展示了运行结果。表示两个线程运行都正常。
在大多数情况下。锁的申请都是非公平的。也就是说系统只是会从等待锁的队列里随机挑选一个，所以不能保证其公平性。但是公平锁的实现成本很高，性能也相对低下。因此如果没有特别要求，也不需要使用公平锁。
对上边ReentrantLock几个重要的方法整理如下。
- lock():获得锁，如果锁已经被占用，则等待。
- lockInterruptibly(): 获得锁，但优先响应中断。
- tryLock():尝试获得锁，如果成功，返回true，失败返回false。该方法不等待，立即返回
- tryLock(long time,TimeUnit unit)，在给定时间内尝试获得锁
- unlock(): 释放锁。注：ReentrantLock的锁释放一定要在finally中处理，否则可能会产生严重的后果。
Condition条件
Conditon和ReentrantLock的组合可以让线程在合适的时间等待，或者在某一个特定的时间得到通知，继续执行。在Condition中，用await()替换wait()，用signal()替换notify()，用signalAll()替换notifyAll()，传统线程的通信方式，Condition都可以实现，这里注意，Condition是被绑定到Lock上的，要创建一个Lock的Condition必须用newCondition()方法。
- await:当前线程进入等待状态，直到被通知（signal OR signalAll）或者被中断时，当前线程进入运行状态，从await()返回；
- awaitUninterruptibly:当前线程进入等待状态，直到被通知，对中断不做响应；
- awaitNanos(long nanosTimeout):在
await()的返回条件基础上增加了超时响应，返回值表示当前剩余的时间，如果在nanosTimeout之前被唤醒，返回值 = nanosTimeout - 实际消耗的时间，返回值 <= 0表示超时； - boolean await(long time, TimeUnit unit):同样是在
await()的返回条件基础上增加了超时响应，与上一接口不同的是可以自定义超时时间单位； 返回值返回true/false，在time之前被唤醒，返回true，超时返回false。 - boolean awaitUntil(Date deadline)：当前线程进入等待状态直到将来的指定时间被通知，如果没有到指定时间被通知返回true，否则，到达指定时间，返回false；
- signal()：唤醒一个等待在Condition上的线程
- signalAll()：唤醒等待在Condition上所有的线程。
使用案例如下
public class ConditionDemo { static class NumberWrapper { public int value = 1; } public static void main(String[] args) { //初始化可重入锁 final Lock lock = new ReentrantLock(); //第一个条件当屏幕上输出到3 final Condition reachThreeCondition = lock.newCondition(); //第二个条件当屏幕上输出到6 final Condition reachSixCondition = lock.newCondition(); //NumberWrapper只是为了封装一个数字，一边可以将数字对象共享，并可以设置为final //注意这里不要用Integer, Integer 是不可变对象 final NumberWrapper num = new NumberWrapper(); //初始化A线程 Thread threadA = new Thread(new Runnable() { @Override public void run() { //需要先获得锁 lock.lock(); try { System.out.println("threadA start write"); //A线程先输出前3个数 while (num.value <= 3) { System.out.println(num.value); num.value++; } //输出到3时要signal，告诉B线程可以开始了 reachThreeCondition.signal(); } finally { lock.unlock(); } lock.lock(); try { //等待输出6的条件 while(num.value <= 6) { reachSixCondition.await(); } System.out.println("threadA start write"); //输出剩余数字 while (num.value <= 9) { System.out.println(num.value); num.value++; } } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } } }); Thread threadB = new Thread(new Runnable() { @Override public void run() { try { lock.lock(); while (num.value <= 3) { //等待3输出完毕的信号 reachThreeCondition.await(); } } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } try { lock.lock(); //已经收到信号，开始输出4，5，6 System.out.println("threadB start write"); while (num.value <= 6) { System.out.println(num.value); num.value++; } //4，5，6输出完毕，告诉A线程6输出完了 reachSixCondition.signal(); } finally { lock.unlock(); } } }); //启动两个线程 threadB.start(); threadA.start(); } }
结果如下
这样看来，Condition和传统的线程通信没什么区别，Condition的强大之处在于它可以为多个线程间建立不同的Condition，下面引入API中的一段代码，加以说明。
class BoundedBuffer { final Lock lock = new ReentrantLock();//锁对象 final Condition notFull = lock.newCondition();//写线程条件 final Condition notEmpty = lock.newCondition();//读线程条件 final Object[] items = new Object[100];//缓存队列 int putptr/*写索引*/, takeptr/*读索引*/, count/*队列中存在的数据个数*/; public void put(Object x) throws InterruptedException { lock.lock(); try { while (count == items.length)//如果队列满了 notFull.await();//阻塞写线程 items[putptr] = x;//赋值 if (++putptr == items.length) putptr = 0;//如果写索引写到队列的最后一个位置了，那么置为0 ++count;//个数++ notEmpty.signal();//唤醒读线程 } finally { lock.unlock(); } } public Object take() throws InterruptedException { lock.lock(); try { while (count == 0)//如果队列为空 notEmpty.await();//阻塞读线程 Object x = items[takeptr];//取值 if (++takeptr == items.length) takeptr = 0;//如果读索引读到队列的最后一个位置了，那么置为0 --count;//个数-- notFull.signal();//唤醒写线程 return x; } finally { lock.unlock(); } } }
这个示例中BoundedBuffer是一个固定长度的集合，这个在其put操作时，如果发现长度已经达到最大长度，那么要等待notFull信号才能继续put，如果得到notFull信号会像集合中添加元素，并且put操作会发出notEmpty的信号，而在其take方法中如果发现集合长度为空，那么会等待notEmpty的信号，接受到notEmpty信号才能继续take，同时如果拿到一个元素，那么会发出notFull的信号。
信号量(Semaphore)
信号量(Semaphore)为多线程协作提供了更为强大的控制用法。无论是内部锁Synchronized还是ReentrantLock，一次都只允许一个线程访问资源，而信号量可以多个线程访问同一资源。Semaphore是用来保护一个或者多个共享资源的访问，Semaphore内部维护了一个计数器，其值为可以访问的共享资源的个数。一个线程要访问共享资源，先获得信号量，如果信号量的计数器值大于1，意味着有共享资源可以访问，则使其计数器值减去1，再访问共享资源。如果计数器值为0,线程进入休眠。当某个线程使用完共享资源后，释放信号量，并将信号量内部的计数器加1，之前进入休眠的线程将被唤醒并再次试图获得信号量。
信号量的UML的类图如下，可以看出和ReentrantLock一样，Semaphore也包含了sync对象，sync是Sync类型；而且，Sync是一个继承于AQS的抽象类。Sync包括两个子类："公平信号量"FairSync 和 "非公平信号量"NonfairSync。sync是"FairSync的实例"，或者"NonfairSync的实例"；默认情况下，sync是NonfairSync(即，默认是非公平信号量)
信号量主要提供了以下构造函数
Semaphore（int num） Semaphore（int num，boolean how）
这里，num指定初始许可计数。因此，它指定了一次可以访问共享资源的线程数。如果是1，则任何时候只有一个线程可以访问该资源。默认情况下，所有等待的线程都以未定义的顺序被授予许可。通过设置how为true，可以确保等待线程按其请求访问的顺序被授予许可。信号量的主要逻辑方法如下
// 从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。 void acquire() // 从此信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞，或者线程已被中断。 void acquire(int permits) // 从此信号量中获取许可，在有可用的许可前将其阻塞。 void acquireUninterruptibly() // 从此信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞。 void acquireUninterruptibly(int permits) // 返回此信号量中当前可用的许可数。 // 释放一个许可，将其返回给信号量。 void release() // 释放给定数目的许可，将其返回到信号量。 // 仅在调用时此信号量存在一个可用许可，才从信号量获取许可。 boolean tryAcquire() // 仅在调用时此信号量中有给定数目的许可时，才从此信号量中获取这些许可。 boolean tryAcquire(int permits) // 如果在给定的等待时间内此信号量有可用的所有许可，并且当前线程未被中断，则从此信号量获取给定数目的许可。 boolean tryAcquire(int permits, long timeout, TimeUnit unit) // 如果在给定的等待时间内，此信号量有可用的许可并且当前线程未被中断，则从此信号量获取一个许可。
实例如下：这里我们模拟10个人去银行存款，但是该银行只有两个办公柜台，有空位则上去存钱，没有空位则只能去排队等待。最后输出银行总额
public class SemaphoreThread { private int customer; public SemaphoreThread() { customer = 0; } /** * 银行存钱类 */ class Bank { private int account = 100; public int getAccount() { return account; } public void save(int money) { account += money; } } /** * 线程执行类，每次存10块钱 */ class NewThread implements Runnable { private Bank bank; private Semaphore semaphore; public NewThread(Bank bank, Semaphore semaphore) { this.bank = bank; this.semaphore = semaphore; } @Override public void run() { int tempCustomer = customer++; if (semaphore.availablePermits() > 0) { System.out.println("客户" + tempCustomer + "启动，进入银行,有位置立即去存钱"); } else { System.out.println("客户" + tempCustomer + "启动，进入银行,无位置，去排队等待等待"); } try { semaphore.acquire(); bank.save(10); System.out.println(tempCustomer + "银行余额为：" + bank.getAccount()); Thread.sleep(1000); System.out.println("客户" + tempCustomer + "存钱完毕，离开银行"); semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); } } } /** * 建立线程，调用内部类，开始存钱 */ public void useThread() { Bank bank = new Bank(); // 定义2个新号量 Semaphore semaphore = new Semaphore(2); // 建立一个缓存线程池 ExecutorService es = Executors.newCachedThreadPool(); // 建立10个线程 for (int i = 0; i < 10; i++) { // 执行一个线程 es.submit(new Thread(new NewThread(bank, semaphore))); } // 关闭线程池 es.shutdown(); // 从信号量中获取两个许可，并且在获得许可之前，一直将线程阻塞 semaphore.acquireUninterruptibly(2); System.out.println("到点了，工作人员要吃饭了"); // 释放两个许可，并将其返回给信号量 semaphore.release(2); } public static void main(String[] args) { SemaphoreThread test = new SemaphoreThread(); test.useThread(); } }
读写锁ReentrantReadWriteLock
ReentrantReadWriteLock是Lock的另一种实现方式，我们已经知道了ReentrantLock是一个排他锁，同一时间只允许一个线程访问，而ReentrantReadWriteLock允许多个读线程同时访问（也就是读操作），但不允许写线程和读线程、写线程和写线程同时访问。约束如下
- 读—读不互斥：读与读之间不阻塞
- 读—写：读阻塞写，写也会阻塞读
- 写—写：写写阻塞
相对于排他锁，提高了并发性。在实际应用中，大部分情况下对共享数据（如缓存）的访问都是读操作远多于写操作，这时ReentrantReadWriteLock能够提供比排他锁更好的并发性和吞吐量。
看一下官方案例
lass CachedData { Object data; volatile boolean cacheValid; final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock(); public void processCachedData() { rwl.readLock().lock();//1 if (!cacheValid) { // Must release read lock before acquiring write lock rwl.readLock().unlock();//2 rwl.writeLock().lock();//3 try { // Recheck state because another thread might have,acquired write lock and changed state before we did. if (!cacheValid) { data = ... cacheValid = true; } // 在释放写锁之前通过获取读锁降级写锁(注意此时还没有释放写锁) rwl.readLock().lock();//4 } finally { // 释放写锁而此时已经持有读锁 rwl.writeLock().unlock();//5 } } try { use(data); } finally { rwl.readLock().unlock();//6 } } }
- 多个线程同时访问该缓存对象时，都加上当前对象的读锁，之后其中某个线程优先查看data数据是否为空。【加锁顺序序号：1 】
- 当前查看的线程，如果发现没有值则释放读锁，然后立即加上写锁，准备写入缓存数据。（进入写锁的前提是当前没有其他线程的读锁或者写锁）【加锁顺序序号：2和3 】
- 为什么还会再次判断是否为空值（!cacheValid）是因为第二个、第三个线程获得读的权利时也是需要判断是否为空，否则会重复写入数据。
- 写入数据后先进行读锁的降级后再释放写锁。【加锁顺序序号：4和5】
- 最后数据数据返回前释放最终的读锁。【加锁顺序序号：6 】
如果不使用锁降级功能，如先释放写锁，然后获得读锁，在这个get过程中，可能会有其他线程竞争到写锁 或者是更新数据 则获得的数据是其他线程更新的数据，可能会造成数据的污染，即产生脏读的问题
public class ReadAndWriteLock {
private static ReentrantLock lock = new ReentrantLock();
private static ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
private static Lock readLock = readWriteLock.readLock();
private static Lock writeLock = readWriteLock.writeLock();
public ReadAndWriteLock setValue(int value) {
this.value = value;
return this;
}
private int value;
public Object handleRead(Lock lock) throws InterruptedException {
try {
//模拟读操作
lock.lock();
System.out.println("thread:" + Thread.currentThread().getId() + " value:" + value);
Thread.sleep(1000);
return value;
} finally {
lock.unlock();
}
}
public Object handleWrite(Lock lock, int index) throws InterruptedException {
try {
//模拟写操作
lock.lock();
value = index;
Thread.sleep(1000);
System.out.println("thread:" + Thread.currentThread().getId() + " value:" + value);
return value;
} finally {
lock.unlock();
}
}
public static void main(String[] args) throws InterruptedException {
final ReadAndWriteLock demo = new ReadAndWriteLock();
demo.setValue(0);
Runnable readRunnable = new Runnable() {
@Override
public void run() {
try {
//读锁
demo.handleRead(readLock);
//可重入锁
//demo.handleRead(lock);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
Runnable writeRunnable = new Runnable() {
@Override
public void run() {
try {
//写锁
demo.handleWrite(readLock, (int) (Math.random() * 1000));
//可重入锁
//demo.handleWrite(lock, (int) (Math.random() * 1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
ExecutorService exec = new ThreadPoolExecutor(0, 200,
0, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
;
long startTime = System.currentTimeMillis();
for (int i = 0; i < 18; i++) {
exec.execute(readRunnable);
}
for (int i = 0; i < 18; i++) {
exec.execute(writeRunnable);
}
exec.shutdown();
exec.awaitTermination(60, TimeUnit.MINUTES);
long endTime = System.currentTimeMillis(); //获取结束时间
System.out.println("程序运行时间： " + (endTime - startTime) + "ms");
}
}
在这里读线程完全并行，而写会阻塞读。程序执行时间如下
将上述案例中的读写锁改成可重入锁，即将第行代码注释掉那么所有的读和写线程都必须相互等待，程序执行时间如下所示
倒计时器：CountDownLatch
CountDownLatch是java1.5版本之后util.concurrent提供的工具类。这里简单介绍一下CountDownLatch，可以将其看成是一个计数器，await()方法可以阻塞至超时或者计数器减至0，其他线程当完成自己目标的时候可以减少1，利用这个机制我们可以将其用来做并发。 比如有一个任务A，它要等待其他4个任务执行完毕之后才能执行，此时就可以利用CountDownLatch来实现这种功能了。
CountDownLatch类只提供了一个构造器，该构造器接受一个整数作为参数，即当前这个计数器的计数个数 。
public CountDownLatch(int count) { }; //参数count为计数值
使用场景：比如对于马拉松比赛，进行排名计算，参赛者的排名，肯定是跑完比赛之后，进行计算得出的，翻译成Java识别的预发，就是N个线程执行操作，主线程等到N个子线程执行完毕之后，在继续往下执行。
public class CountDownLatchTest { public static void main(String[] args){ int threadCount = 10; final CountDownLatch latch = new CountDownLatch(threadCount); for (int i = 0; i < threadCount; i++) { new Thread(new Runnable() { @Override public void run() { System.out.println("线程" + Thread.currentThread().getId() + "开始出发"); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("线程" + Thread.currentThread().getId() + "已到达终点"); latch.countDown(); } }).start(); } try { latch.await(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("10个线程已经执行完毕！开始计算排名"); } }
结果如下
线程12开始出发
线程14开始出发
线程15开始出发
线程17开始出发
线程13开始出发
线程16开始出发
线程18开始出发
线程19开始出发
线程20开始出发
线程21开始出发
线程16已到达终点
线程13已到达终点
线程19已到达终点
线程18已到达终点
线程17已到达终点
线程14已到达终点
线程15已到达终点
线程12已到达终点
线程21已到达终点
线程20已到达终点
10个线程已经执行完毕！开始计算排名
CountDownLatch在并行化应用中也是比较常用。常用的并行化框架OpenMP中也是借鉴了这种思想。比如有这样的一个需求，在你淘宝订单的时候，这笔订单可能还需要查，用户信息，折扣信息，商家信息，商品信息等，用同步的方式（也就是串行的方式）流程如下。
设想一下这5个查询服务，平均每次消耗100ms，那么本次调用至少是500ms，我们这里假设，在这个这五个服务其实并没有任何数据依赖，谁先获取谁后获取都可以，那么我们可以想办法并行化这五个服务。
这里可以使用CountDownLatch来实现这个效果。
public class CountDownDemo { private static final int CORE_POOL_SIZE = 4; private static final int MAX_POOL_SIZE = 8; private static final long KEEP_ALIVE_TIME = 5L; private final static int QUEUE_SIZE = 1600; protected final static ExecutorService THREAD_POOL = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, new LinkedBlockingQueue<>(QUEUE_SIZE)); public static void main(String[] args) throws InterruptedException { // 新建一个为5的计数器 CountDownLatch countDownLatch = new CountDownLatch(5); OrderInfo orderInfo = new OrderInfo(); THREAD_POOL.execute(() -> { System.out.println("当前任务Customer,线程名字为:" + Thread.currentThread().getName()); orderInfo.setCustomerInfo(new CustomerInfo()); countDownLatch.countDown(); }); THREAD_POOL.execute(() -> { System.out.println("当前任务Discount,线程名字为:" + Thread.currentThread().getName()); orderInfo.setDiscountInfo(new DiscountInfo()); countDownLatch.countDown(); }); THREAD_POOL.execute(() -> { System.out.println("当前任务Food,线程名字为:" + Thread.currentThread().getName()); orderInfo.setFoodListInfo(new FoodListInfo()); countDownLatch.countDown(); }); THREAD_POOL.execute(() -> { System.out.println("当前任务Tenant,线程名字为:" + Thread.currentThread().getName()); orderInfo.setTenantInfo(new TenantInfo()); countDownLatch.countDown(); }); THREAD_POOL.execute(() -> { System.out.println("当前任务OtherInfo,线程名字为:" + Thread.currentThread().getName()); orderInfo.setOtherInfo(new OtherInfo()); countDownLatch.countDown(); }); countDownLatch.await(1, TimeUnit.SECONDS); System.out.println("主线程：" + Thread.currentThread().getName()); } }
建立一个线程池(具体配置根据具体业务，具体机器配置)，进行并发的执行我们的任务(生成用户信息，菜品信息等)，最后利用await方法阻塞等待结果成功返回。
循环栅栏CyclicBarrier
字面意思循环栅栏，栅栏就是一种障碍物。这里就是内存屏障。通过它可以实现让一组线程等待至某个状态之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后，CyclicBarrier可以被重用。CyclicBarrier比CountDownLatch 功能更强大一些，CyclicBarrier可以接受一个参数作为barrierAction。所谓barrierAction就是当计算器一次计数完成后，系统会执行的动作。CyclicBarrier强调的是n个线程，大家相互等待，只要有一个没完成，所有人都得等着。（这种思想在高性能计算最为常见，GPU计算中关于也有类似内存屏障的用法）。构造函数如下，其中parties表示计数总数，也就是参与的线程总数。
public CyclicBarrier(int parties, Runnable barrierAction) { } public CyclicBarrier(int parties) { }
案例10个人去旅行,规定达到一个地点后才能继续前行.代码如下
class CyclicBarrierWorker implements Runnable { private int id; private CyclicBarrier barrier; public CyclicBarrierWorker(int id, final CyclicBarrier barrier) { this.id = id; this.barrier = barrier; } @Override public void run() { try { Thread.sleep(Math.abs(new Random().nextInt()%10000)); System.out.println(id + " th people wait"); barrier.await(); // 大家等待最后一个线程到达 } catch (InterruptedException | BrokenBarrierException e) { e.printStackTrace(); } } } public class TestCyclicBarrier { public static void main(String[] args) { int num = 10; CyclicBarrier barrier = new CyclicBarrier(num, new Runnable() { @Override public void run() { System.out.println("go on together!"); } }); for (int i = 1; i <= num; i++) { new Thread(new CyclicBarrierWorker(i, barrier)).start(); } } }
从上面输出结果可以看出，每个线程执行自己的操作之后，就在等待其他线程执行操作完毕。当所有线程线程执行操作完毕之后，所有线程就继续进行后续的操作了。
评论暂时关闭