一、ReentrantLock 类
1.1 什么是reentrantlock
java.util.concurrent.lock 中的 Lock 框架是锁定的一个抽象,它允许把锁定的实现作为 Java 类,而不是作为语言的特性来实现。这就为 Lock 的多种实现留下了空间,各种实现可能有不同的调度算法、性能特性或者锁定语义。 ReentrantLock 类实现了 Lock ,它拥有与 synchronized 相同的并发性和内存语义,但是添加了类似锁投票、定时锁等候和可中断锁等候的一些特性。此外,它还提供了在激烈争用情况下更佳的性能。(换句话说,当许多线程都想访问共享资源时,JVM 可以花更少的时候来调度线程,把更多时间用在执行线程上。)
reentrant 锁意味着什么呢? 简单来说,它有一个与锁相关的获取计数器,如果拥有锁的某个线程再次得到锁,那么获取计数器就加1,然后锁需要被释放两次才能获得真正释放。这模仿了 synchronized 的语义;如果线程进入由线程已经拥有的监控器保护的 synchronized 块,就允许线程继续进行,当线程退出第二个(或者后续) synchronized 块的时候,不释放锁,只有线程退出它进入的监控器保护的第一个 synchronized 块时,才释放锁。
1.2 ReentrantLock与 synchronized的比较
相同: ReentrantLock提供了synchronized类似的功能和内存语义。
不同:
(1)ReentrantLock功能性方面更全面,比如时间锁等候,可中断锁等候,锁投票等,因此更有扩展性。在多个条件变量和高度竞争锁的地方,用ReentrantLock更合适,ReentrantLock还提供了Condition,对线程的等待和唤醒等操作更加灵活,一个ReentrantLock可以有多个Condition实例,所以更有扩展性。
(2)ReentrantLock 的性能比synchronized会好点。
(3)ReentrantLock提供了可轮询的锁请求,他可以尝试的去取得锁,如果取得成功则继续处理,取得不成功,可以等下次运行的时候处理,所以不容易产生死锁,而synchronized则一旦进入锁请求要么成功,要么一直阻塞,所以更容易产生死锁。
1.3 ReentrantLock扩展的功能
1.3.1 实现可轮询的锁请求
在内部锁中,死锁是致命的——唯一的恢复方法是重新启动程序,唯一的预防方法是在构建程序时不要出错。而可轮询的锁获取模式具有更完善的错误恢复机制,可以规避死锁的发生。
如果你不能获得所有需要的锁,那么使用可轮询的获取方式使你能够重新拿到控制权,它会释放你已经获得的这些锁,然后再重新尝试。可轮询的锁获取模式,由tryLock()方法实现。此方法仅在调用时锁为空闲状态才获取该锁。如果锁可用,则获取锁,并立即返回值true。如果锁不可用,则此方法将立即返回值false。此方法的典型使用语句如下:
Lock lock = ...;
if (lock.tryLock()) {
try {
// manipulate protected state
} finally {
lock.unlock();
}
} else {
// perform alternative actions
}}
1.3.2 实现可定时的锁请求
当使用内部锁时,一旦开始请求,锁就不能停止了,所以内部锁给实现具有时限的活动带来了风险。为了解决这一问题,可以使用定时锁。当具有时限的活
动调用了阻塞方法,定时锁能够在时间预算内设定相应的超时。如果活动在期待的时间内没能获得结果,定时锁能使程序提前返回。可定时的锁获取模式,由tryLock(long, TimeUnit)方法实现。
1.3.3 实现可中断的锁获取请求
可中断的锁获取操作允许在可取消的活动中使用。lockInterruptibly()方法能够使你获得锁的时候响应中断。
1.4 ReentrantLock不好与需要注意的地方
(1) lock 必须在 finally 块中释放。否则,如果受保护的代码将抛出异常,锁就有可能永远得不到释放!这一点区别看起来可能没什么,但是实际上,它极为重要。忘记在 finally 块中释放锁,可能会在程序中留下一个定时bug,当有一天bug爆炸时,您要花费很大力气才有找到源头在哪。而使用同步,JVM 将确保锁会获得自动释放
(2) 当 JVM 用 synchronized 管理锁定请求和释放时,JVM 在生成线程转储时能够包括锁定信息。这些对调试非常有价值,因为它们能标识死锁或者其他异常行为的来源。 Lock 类只是普通的类,JVM 不知道具体哪个线程拥有 Lock 对象。
二、 条件变量 Condition
条件变量很大一个程度上是为了解决Object.wait/notify/notifyAll难以使用的问题。
条件 (也称为 条件队列 或 条件变量 )为线程提供了一个含义,以便在某个状态条件现在可能为 true 的另一个线程通知它之前,一直挂起该线程(即让其“等待”)。因为访问此共享状态信息发生在不同的线程中,所以它必须受保护,因此要将某种形式的锁与该条件相关联。等待提供一个条件的主要属性是: 以原子方式 释放相关的锁,并挂起当前线程,就像 Object.wait 做的那样。
上述API说明表明条件变量需要与锁绑定,而且多个Condition需要绑定到同一锁上。前面的 Lock中提到,获取一个条件变量的方法是 Lock.newCondition() 。
void await() throws InterruptedException;
void awaitUninterruptibly();
long awaitNanos(long nanosTimeout) throws InterruptedException;
boolean await(long time, TimeUnit unit) throws InterruptedException;
boolean awaitUntil(Date deadline) throws InterruptedException;
void signal();
void signalAll();
以上是 Condition 接口定义的方法, await* 对应于 Object.wait , signal 对应于 Object.notify , signalAll 对应于 Object.notifyAll 。特别说明的是 Condition 的接口改变名称就是为了避免与Object中的 wait/notify/notifyAll 的语义和使用上混淆,因为Condition同样有 wait/notify/notifyAll 方法。
每一个 Lock 可以有任意数据的 Condition 对象, Condition 是与 Lock 绑定的,所以就有 Lock的公平性特性:如果是公平锁,线程为按照FIFO的顺序从 Condition.await 中释放,如果是非公平锁,那么后续的锁竞争就不保证FIFO顺序了。
一个使用Condition实现 生产者消费者 的模型例子如下。
import java.util.Random; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @see ArrayBlockingQueue * @author fansxnet * * @param <T> */ public class ProductQueue<T> { private final T[] items; private final Lock lock; private Condition notFull; private Condition notEmpty; // private int head, tail, count; public ProductQueue(int maxSize, boolean fair) { items = (T[]) new Object[maxSize]; this.lock = new ReentrantLock(fair); this.notFull = this.lock.newCondition(); this.notEmpty = this.lock.newCondition(); } public ProductQueue() { this(10,false); } public void put(T t) throws InterruptedException { lock.lockInterruptibly(); try { while (count == getCapacity()) { notFull.await(); } items[tail] = t; if (++tail == getCapacity()) { tail = 0; } ++count; notEmpty.signalAll(); } finally { lock.unlock(); } } public T take() throws InterruptedException { lock.lockInterruptibly(); try { while (count == 0) { notEmpty.await(); } T ret = items[head]; items[head] = null;// GC // if (++head == getCapacity()) { head = 0; } --count; notFull.signalAll(); return ret; } finally { lock.unlock(); } } public int getCapacity() { return items.length; } public int size() { lock.lock(); try { return count; } finally { lock.unlock(); } } public static void main(String[] args) throws InterruptedException { ExecutorService pool = Executors.newFixedThreadPool(10); final ProductQueue<Integer> queue = new ProductQueue<Integer>(); final CountDownLatch cdOrder = new CountDownLatch(1); for (int i = 0; i < 5; i++) { pool.submit(new Runnable() { public void run() { try { cdOrder.await(); System.out.println("start put "); while (true) { int c = new Random().nextInt(100); System.out.println("put" + c); queue.put(c); } } catch (InterruptedException e) { e.printStackTrace(); } } }); pool.submit(new Runnable() { public void run() { try { cdOrder.await(); System.out.println("start take "); while (true) { System.out.println("take" + queue.take()); } } catch (InterruptedException e) { e.printStackTrace(); } } }); } System.out.println("working"); cdOrder.countDown(); pool.shutdown(); } }
在这个例子中消费 take() 需要 队列不为空,如果为空就挂起( await() ),直到收到 notEmpty的信号;生产 put() 需要队列不满,如果满了就挂起( await() ),直到收到 notFull 信号。
参考
Java多线程(九)之ReentrantLock与Condition