《java并发编程实践》第5章要点总结（阻塞队列、闭锁、关卡）

读完本文你将对以下几个同步异步的知识点有所了解：

显式隐式迭代器和

ConcurrentMidificationException

并发容器

ConcurrentHashMap

和

CopyOnWriteArrayList

阻塞队列和生产者-消费者模式

闭锁和关卡

显式隐式迭代器和

ConcurrentMidificationException

首先看下面两个方法：

public static Object getLast(Vector list){
int lastIndex = list.size() - 1;
return list.get(lastIndex);
}

public static void deleteLast(Vector list){
int lastIndex = list.size() - 1;
list.remove(lastIndex);
}

很显然，如果两个线程同时访问，返回的size大小相同，但是线程2首先获得时间片，进入

remove

方法，删除最后一个元素，此时线程1再调用

get

方法会报出数组下标越界的错误。

根据客户端加锁的解决方法， 我们在两个方法内都加上如下的对象锁，使得对同样的

list

无法同时进行这两个操作。

public static Object getLast(Vector list){
synchronized(list){
...
}
}

好像没什么问题了，下面我们迭代看看。

for(int i = 0; i < list.size(); i++){
doSomething(list.get(i));
}

这种迭代在单线程环境下毫无问题，但是在多线程并发修改

list

的时候，就有问题了。和上面的例子是同样的道理，并发的删除和迭代操作不具有原子性，一旦删除先发生，迭代肯定会出错。

这并不能算线程不安全，相反异常恰好使其保证了规约的一致性。但是抛出异常毕竟不是人们期望的。

可以通过加锁的方式对迭代进行保护，但是如果数据量大，迭代需要的时间很长，长时间阻塞其他线程的访问会对性能有巨大的影响。

接下来步入正题：

对集合的遍历标准方式是通过

Iterator

，无论是显式调用还是java5.0引入的

for-each

语法，但是同步容器的迭代器是“及时失败”的，当它察觉容器在迭代开始后被修改，会抛出一个未检查的

ConcurrentModificationException

异常。

注意：这个异常不只是出现在多线程中，单线程中也可能出现，如果遍历开始后，直接在集合中删除元素而没有通过

Iterator.remove()

方法删除，就会抛出此异常。

还有一些隐藏的迭代器，比如字符拼接操作，例如：

private List mList;

System.out.println("It also means a iterator" + mList);

实际上字符拼接在经过编译转换后会成为调用

StringBuilder.append(Object)

，它会调用容器的

toString

方法，而

toString

的实现基于遍历容器中的元素，所以也算是迭代器的使用。

另外

hashCode

和

equals

，

containsAll

等方法也都会对容器进行迭代，所以也可能引起

ConcurrentModificationException

。

如何在迭代期间避免出现上述

Exception

，数据少的情况下可以考虑使用锁，而另一种办法是复制容器，例如

List.addAll()

方法，将所有元素复制一份进入新的容器，但是复制容器的性能开销也很大，具体情况具体测试。

并发容器

ConcurrentHashMap

和

CopyOnWriteArrayList

ConcurrentHashMap

ConcurrentHashMap

和

hashMap

都是哈希表，但是使用了完全不同的锁策略。

hashMap

本身是线程不安全的，如果要达到线程安全必须上锁，而使用的是一个公共锁同步每个方法，保证同一时刻只有一个线程访问，可见效率是极低的，并行与串行几乎没区别。

而

ConcurrentHashMap

使用的是分离锁，分离锁就是针对相互独立的变量分成若干加锁块的集合，而不是所有变量加同一个锁，分离锁允许任意数量的读线程可以并发的访问

Map

，有限数量的写线程也可以并发的修改

Map

，并且读写线程可以同时作用，带来高吞吐量的同时又几乎没有损失单个线程的访问性能。

并且，关键的一点是，在

ConcurrentHashMap

中使用迭代，不会抛出

ConcurrentModificationException

，其迭代器可以容许并发修改，并且可以（不保证）在迭代器创建以后，感应修改。

总的来说，相比于其他同步Map，

ConcurrentHashMap

有众多优势，几乎没有劣势，唯一一点就是当程序需要在独占访问中加锁，比如说对元素进行若干次迭代并且要求元素的出现顺序相同的情况下，

ConcurrentHashMap

无法胜任，其他时候还是使用它吧。

PS：

ConcurrentHashMap

除了不支持对客户端加锁而产生的原子操作，其他的例如“缺少即加入”，“相等便移除”等原子操作都是支持的。

CopyOnWriteArrayList

用于替代同步

List

，又称“写入时复制”，其内部原理是：

只要有效的不可变对象被正确发布，访问它就不再需要同步，并且每次修改的时候，会创建并重新发布一个新的容器拷贝。其迭代器保留了一个底层数组的引用，这个数组作为迭代器的起点，永远不会被修改。该容器返回的迭代器不会抛出

ConcurrentModificationException

，并且返回的元素严格与迭代器创建时一致。

显然，每次容器内容改变时复制基础数组都有一定的开销，但是当迭代的频率远远高于对容器的修改频率的时候，这种开销可以被近似忽略。观察者模式就是个很好的例子，当模型创建好了之后，有消息时会遍历观察者数组通知每个观察者，此时修改的频率很低，远远低于通知发送的频率。

阻塞队列和生产者-消费者模式

阻塞队列顾名思义，是一个支持长时间阻塞的队列，普通的队列都是FIFO，并且

take

和

put

操作都是瞬时的，一旦没有元素，那么

take

操作经过判断以后结束，而阻塞队列的

take

操作另开线程，通过无限循环查询队列是否有元素，一旦有元素就取出，否则继续堵塞。

生产者-消费者模式指的是将工作的产生者和工作的执行者分离，不再关心二者之间的耦合关系，将更多的精力放在工作的实现上。生产者不会将工作立刻给予消费者，而是放在

to-do

的清单中，消费者通过需要从中取出工作。

该模式是围绕阻塞队列设计的，生产者将数据放入队列，消费者取出数据操作。二者根本没有管对方是谁，只要独立完成自己的工作即可，这利于抽象的设计。线程池和工作队列的模型就是依赖于该模式。

常见的阻塞队列有

LinkedBlockingQueue

和

ArrayBlockingQueue

，这两个是FIFO队列，

PriorityBlockingQueue

是优先级队列，会根据数据的自然顺序进行排列（前提是数据已实现Comparable），也可以使用自己的

Comparator

。

SynchronousQueue

不是一个真正的队列，它维护着排队线程的清单，这些线程用于执行队列的操作，它会即时传递消息，不会将操作放在某处等待运行，所以能获得即时的反馈，适合在消费者充足的情况下使用。

以下是一个桌面搜索应用程序的生产者和消费者模型：

public static FileCrawler implements Runnable { //生产者
private final BlockingQueue<File> fileQueue; //阻塞队列
private final FileFilter fileFilter;
private final File root;
...

public void run() {
try {
crawl(root);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}

private void crawl(File root) throws InterruptedException{
File[] entries = root.listFiles(fileFilter);
if(entries != null){
for (File entry : entries)
if(entry.isDirectory())
crawl(entry);
else if(!alreadyIndexed(entry))
fileQueue.put(entry);
}
}

public class Indexer implements Runnable { //消费者
private final BlockingQueue<File> queue;
//to-do constructor
public void run(){
try {
while(true)
indexFile(queue.take());
} catch (InterruptedException e){
Thread.currentThread().interrupt();
}
}
}
}

public static void startIndexing(File[] roots){
BlockingQueue<File> queue = new LinkedBlockingQueue<File>(BOUND);
FileFilter filter = new FileFilter(){
public boolean accept(File file) {return true;}
}
for(File root : roots)
new Thread(new FileCrawler(queue, filter, root)).start();
for(int i=0; i<N_CONSUMERS; i++)
new Thread(new Indexer(queue)).start;
}

生产者负责将每个文件提交到阻塞队列中，消费者通过无限循环从队列中取出文件操作，代码中为每个生产者和消费者都创建了单独的线程，实现了并发操作。

最后提一下双端队列和窃取工作：

双端队列

Deque

允许高效地在头尾分别进行插入和移除操作，与其关联的是窃取工作的模式。

在生产者-消费者模式中，消费者共用一个工作队列，而窃取工作中，每个消费者都拥有自己的双端队列，一个消费者在完成它自己队列的任务后，可以窃取其他消费者队列末端的任务。由于避免了对一个共享队列的竞争，同时在末尾窃取又减少了与其他消费者的竞争，所以窃取工作的效率很高。

适用于双端队列的场景：当运行到一个任务时，可能会识别出更多的任务，比如说：Web处理一个页面的时候，通常会有更多的页面待搜索，以及垃圾回收时对堆做标记，并行使用窃取工作。

闭锁和关卡

闭锁

闭锁可以延迟线程的进度直到线程终止的状态，在未达到终止状态之前，所有的线程都不能通过这个节点，一旦达到终止，允许线程都通过，并且闭锁的状态不会再改变。以下几个场景用于闭锁：

确保一个计算不会执行，直到它所需要的资源都已经初始化完毕。

确保一个服务不会开始，直到它依赖的服务都已经开始。

在游戏中确保所有玩家准备就绪，才能开始游戏

CountDownLatch

这是闭锁的实现之一，允许一个或多个线程等待一个时间集的发生。首先初始化为一个正数，表示需要等待的事件数，

countDown

方法用来对计数器做自减操作，表示一个事件已经发生，

await

方法则等待该计数器为0，若非0则等待，一直处于阻塞状态，直到线程中断或者超时。

下面这段代码实现了多个线程在同一时刻开启，并且计算最后一个线程结束所到的时间，返回总的运行时间。

public static Test(){
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++){
new Thread(){
public void run(){
try{
startGate.await(); //startGate使得线程等待直到countDown统一开启
try{
task.run();
} finally {
endGate.countDown(); //endGate每次线程执行完记录一次
}
} catch(InterruptedException e){}
}
}.start;
}
long start = System.nanoTime();
startGate.countDown();
endGate.await; //最后一个线程结束后该方法调用结束
longt end = System.nanoTime();
return end - start;
}
}

为何不一开始就启动所有线程呢？这是为了计算在n倍线程并发下执行一个任务的时间。迭代有先后，这会导致每个线程的开启时间不一致，先开启的线程在竞争中就有“领先优势”，并且在线程增加或减少的时候竞争度在不断改变，测量的时间就不是那么准确。

FutureTask

FutureTask

也是闭锁的实现之一，其描述了一个抽象的可携带结果的计算。其计算是通过

Callable

实现的，这个

Callable

等价于一个可携带结果的

Runnable

，并且有等待、运行、完成三种状态。如果计算完成，调用

Future.get

可以立刻获得返回结果，否则会阻塞直到任务进入完成状态，然后返回结果或者抛出异常。

FutureTask

内部已经为我们切换了线程，将返回结果从计算线程传给需要该结果的线程。可见Executor框架就是通过它来完成异步任务的。

下面举个例子：

public class PreLoad {
private final FutureTask<ProductInfo> future =
new FutureTask<ProductInfo>(new Callable<ProductInfo>(){
public ProductInfo call() throws DataLoadException{
return loadProductInfo();
}
});
private final Thread t = new Thread(future);
public void start(){ t.start(); }
public ProductInfo get() throws DataLoadException, InterruptedException {
try{
return future.get();
} catch (ExecutionException e){
Throwable cause = e.getCause();
if(cause instanceof DataLoadException)
throw (DataLoadException)cause;
else
throw launderThrowable(cause);
}
}
}

由于

Callable

抛出的所有异常都被封装在了

ExecutionException

中，所以需要对其进行详细的归类，使得抛出的异常更容易处理。

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

管理一个有效的许可集，许可的初始量通过构造函数传给信号量，活动通过

acquire

获取可用的信号量，并在使用之后释放，如果没有可用的信号量，那么

acquire

会堵塞，直到获得、中断或者超时。

release

方法向信号量返回一个许可。

例如下面的例子，利用信号量控制容器的大小：

public class BoundedHashSet<T> {
private final Set<T> set;
private final Semaphore sem;
public BoundedHashSet(int bound){
this.set = Collections.synchronizedSet(new HashSet<T>());
sem = new Semaphore(bound);
}

public boolean add(T o) throws InterruptedException {
sem.acquire();
boolean wasAdded = false;
try{
wasAdded = set.add(o);
return wasAdded;
} finally{
if(!wasAdded)
sem.release();
}
}

public boolean remove (Object o){
boolean removed = set.remove(o);
if(removed)
sem.release();
return removed;
}
}

关卡

关卡类似于闭锁，不同点在于关卡中，所有的线程必须同时到达关卡点，才能继续处理。闭锁等待的是事件，而关卡等待的是其他线程。此外闭锁达到最终状态时不能够重置，关卡可以。

关卡的一个实现是

CyclicBarrier

，其将一个问题分成若干个子问题交给各个线程解决，线程完成后调用

await

，阻塞当前线程，一旦

await

使得计数器自增到给定的初始值，则关卡放行。

下面是一个例子：

public class Test1 {

public static void main(String[] args) throws InterruptedException, BrokenBarrierException{

CyclicBarrier barrier = new CyclicBarrier(4);
for(int i=0; i<3; i++){
new Writer(barrier).start();
}
barrier.await();
System.out.println("所有数据均写完！");
}

static class Writer extends Thread{

private CyclicBarrier innerBarrier;

public Writer(CyclicBarrier barrier){
innerBarrier = barrier;
}

public void run(){
try {
Thread.sleep(1000);
System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");
innerBarrier.await();
}catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

}

运行结果：

线程Thread-0写入数据完毕，等待其他线程写入完毕
线程Thread-1写入数据完毕，等待其他线程写入完毕
线程Thread-2写入数据完毕，等待其他线程写入完毕
所有数据均写完！

每个线程运行完就进入等待状态，等到所有线程均进入等待状态时，计数器恰好为4，则关卡放行，执行

await

后面的方法。