synchronizer--JCIP5.5读书笔记
[本文是我对Java Concurrency In Practice 5.5的归纳和总结. 转载请注明作者和出处, 如有谬误, 欢迎在评论中指正. ]
Synchronizers
synchronizer是指那些根据状态协调线程的对象. BlockingQueue就是一个典型的synchronizer: 当队列为空时, 取数据的消费者线程将被阻塞, 直到队列不为空; 当队列满时, 存入数据的生产者线程将被阻塞, 直到队列不满. 除了BlockingQueue之外, 常用的synchronizer还有latch, FutureTask, semaphore, barrier等.
latch
latch相当于一个gate, 当latch到达特定的状态之前, gate是关闭的, 此时所有线程将被阻塞, 只有latch到达了特定状态, 线程才能通过gate.
CountDownLatch是latch的具体实现, CountDownLatch内部维护了一个计数器, 初始化CountDownLatch时需要指定计数器的初始值. 该初始值表示需要等待完成的事件的个数. 每调用一次countDown方法, 表示其中一个事件已经完成, 计数器的值将减一. 当计数器减为0时, gate才会打开:
public class TestHarness {
/**
* 计算nThreads个线程并发执行task所需的时间
*/
public long timeTasks(int nThreads, final Runnable task) throws InterruptedException {
final CountDownLatch startGate = new CountDownLatch(1);
final CountDownLatch endGate = new CountDownLatch(nThreads);
for (int i = 0; i < nThreads; i++) {
Thread t = new Thread() {
public void run() {
try {
// 直到startGate内部计数器减为0时才能从await中唤醒
startGate.await();
try {
task.run();
} finally {
// 完成了一件任务后将endGate减1
endGate.countDown();
}
} catch (InterruptedException ignored) {
}
}
};
t.start();
}
long start = System.nanoTime();
// 将startGate内部的计数器减1, 打开startGate
startGate.countDown();
// 直到endGate内部计数器减为0时才能从await中唤醒
endGate.await();
long end = System.nanoTime();
return end - start;
}
}
FutureTask
FutureTask用于执行任务, 其get方法将返回任务的执行结果. FutureTask常用的构造函数为FutureTask(Callable<V> callable), 使用Callable封装任务. FutureTask对象具有三种状态: 等待运行, 正在运行, 已完成. 当FutureTask对象处于已完成状态时调用get方法, get方法将立即返回计算结果, 否则get方法会阻塞, 直到FutureTask转变为已完成状态. 计算完成, 抛出异常, 或者被取消都会使得FutureTask的状态变为已完成.
FutureTask的常见使用场景是封装一个耗时任务, 然后提前开始计算, 当需要计算结果时, 再调用其get方法, 这样可以减少等待计算完成的时间.
使用FutureTask的例子:
public class Preloader {
private final FutureTask<ProductInfo> future = new FutureTask<ProductInfo>(new Callable<ProductInfo>() {
public ProductInfo call() throws DataLoadException {
// loadProductInfo方法用于加载产品信息, 这是一个耗时操作
return loadProductInfo();
}
});
// FutureTask实现了Runnable接口, 因此可以将future对象放进thread中执行
private final Thread thread = new Thread(future);
// 提供start方法启动线程, 而不是在构造方法中直接启动是为了防止this逃逸
public void start() {
// 启动thread, 执行future
thread.start();
}
// 当需要计算结果时, 就调用get方法获得产品信息的加载结果.
public ProductInfo get() throws DataLoadException, InterruptedException {
try {
// get方法将阻塞, 直到产品信息加载完成, 或者抛出异常
return future.get();
} catch (ExecutionException e) {
Throwable cause = e.getCause();
if (cause instanceof DataLoadException)
throw (DataLoadException) cause;
else
throw launderThrowable(cause);
}
}
public static RuntimeException launderThrowable(Throwable t) {
if (t instanceof RuntimeException)
return (RuntimeException) t;
else if (t instanceof Error)
throw (Error) t;
else
throw new IllegalStateException("Not unchecked", t);
}
}
semaphore--信号量
semaphore用于管理permit, 创建Semaphore对象时, 需要指定permit的最大个数. 调用acquire()方法申请从Semaphore对象中获取一个permit, 如果当前semaphore对象没有可用的permit, 线程将被阻塞, 直到有可用的permit. 调用release()方法将permit放回Semaphore对象. permit不与线程绑定, 一个线程申请的permit, 可以在另一个线程里release. semaphore通常用于实现资源池, 如数据库连接池等. semaphore也可以用于实现有界的集合, 如:
/**
* 有界的set集合
*/
public class BoundedHashSet<T> {
private final Set<T> set;
private final Semaphore sem;
public BoundedHashSet(int bound) {
this.set = Collections.synchronizedSet(new HashSet<T>());
// 设定Semaphore对象中的permit的最大个数
sem = new Semaphore(bound);
}
public boolean add(T o) throws InterruptedException {
// 每次add时就向semaphore对象申请一个permit
sem.acquire();
boolean wasAdded = false;
try {
wasAdded = set.add(o);
return wasAdded;
}
finally {
if (!wasAdded)
// 当申请失败是release permit
sem.release();
}
}
public boolean remove(Object o) {
boolean wasRemoved = set.remove(o);
if (wasRemoved)
// 成功移除后将permit release
sem.release();
return wasRemoved;
}
}
CyclicBarrier
CyclicBarrier允许一组线程互相等待, 直到该组线程全部到达某个公共屏障点. 创建CyclicBarrier时需要指定线程组中线程的数量. 调用CyclicBarrier对象的await方法, 表示当前线程已到达公共屏障点, 然后等待其他线程到达. 当所有线程到达公共屏障点后, CyclicBarrier对象将释放线程组, 然后重置CyclicBarrier对象的状态. 因此CyclicBarrier对象是可以循环使用的. 如果有线程在等待期间超时或者被中断, 该CyclicBarrier对象被视为已损坏, 随后对await方法的调用都要抛出BrokenBarrierException异常.