《Java并发编程实践》笔记2——基础同步类

1.同步容器类复合操作容易出现的问题：

JDK中同步容器类包括两部分：

早期JDK中的Vector和Hashtable；

JDK1.2引入的同步包装类，即由Collections.synchronizedXxx工厂方法创建的容器类。

同步容器类都是线程安全的，它们提供的基本操作都是原子操作，但是对于诸如迭代、导航和缺少即加入的条件运算等复合操作来说，通常需要使用额外的客户端加锁进行保护来确保线程安全。

(1).条件操作：

下面例子演示同步容器类的非线程安全条件运算：

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);
}

上述代码在单线程中没有任何问题，但是如果是多线程并发运行则问题会非常大，例如一个拥有10个元素的Vector，一个线程调用其getLast方法，与此同时另一个线程调用其deleteLast方法，若deleteLast方法早于getLast方法执行，则就会导致ArrayIndexOutOfBoundsException。

修复上述问题方法很简单，使用客户端加锁，例子代码如下：

public static object getLast(Vector list){
synchronized(list){
int lastIndex = list.size() - 1;
return list.get(lastIndex);
}
}
public static void deleteLast(Vector list){
synchronized(list){
int lastIndex = list.size() - 1;
list.remove(lastIndex);
}
}
(2).迭代：

下面例子演示同步容器类的非线程安全迭代：

for(int i = 0; i < vector.size(); i++){
doSomething(vector.get(i));
}
若多个线程并发运行环境下，一个线程调用vector的delete操作，一个线程调用上面的迭代操作，则size和get方法在调用直接vector已经发生变化，因此也会导致ArrayIndexOutOfBoundsException。

修复上述问题方法很简单，同样使用客户端加锁，例子代码如下：

synchronized(vector){
for(int i = 0; i < vector.size(); i++){
doSomething(vector.get(i));
}
}
通过客户端加锁，确保了同步容器复合操作的线程安全，但是却增加了可伸缩性开销，削弱了并发性。

2.容器的ConcurrentModificationException：

容器的并发修改异常ConcurrentModificationException在单线程和多线程都会发生，很多对集合不熟悉的人都可能多次碰到过这个问题，下面详细介绍：

(1).单线程环境下的并发修改异常：

单线程环境下，如果在对一个集合容器进行遍历的同时对集合容器直接进行增删操作，则就会导致并发修改异常，例子代码如下：

for(int i = 0; i < list.size(); i++){  
if(list.get(i) == 2){  
list.remove(i);  
    }  
} 
运行上面例子时就会产生发修改异常ConcurrentModificationException，解决这个问题很简单，遍历时用迭代器增删元素，例子代码如下：

Iterator it = list.iterator();
while(it.hasNext()){  
int value = it.next();
if(value == 2){  
it.remove();  
    }  
} 
(2).多线程环境下的并发修改异常：

上述使用迭代器在遍历集合的时候删除集合元素在单线程可以避免修改异常ConcurrentModificationException，但是在多线程环境仍然会发生修改异常ConcurrentModificationException。

在多线程情况下迭代器发生并发修改异常的原因是同步容器设计时并没有考虑并发修改问题，它们通常通过把修改计数器(modification count)与容器关联起来，如果在迭代期间计数器被修改，在hasNext或next方法就会通过抛出一个ConcurrentModificationException异常的及时失败(fail-fast)方式告知不能支持并发修改。

解决多线程并发修改的方法如下：

A.在所有遍历增删地方都加上synchronized或者使用Collections.synchronizedXxx，虽然能解决问题但是并不推荐，因为增删造成的同步锁可能会阻塞遍历操作。

B.推荐使用ConcurrentHashMap或者CopyOnWriteArrayList等并发容器。

3.并发容器：

同步容器通过对容器的所有状态进行串行访问，从而实现了它们的线程安全，但是这样做的代价是削弱了并发性，当多个线程共同竞争容器级的锁时，吞吐量将会下降。

为了改进同步容器的并发性，JDK1.5引入了并发容器，这些并发容器提供多线程环境下不会抛出并发修改异常的迭代器，常用的并发容器如下：

(1).ConcurrentHashMap：

同步容器使用一个公共锁同步每一个方法，并严格限制只能有一个线程同时访问容器，而ConcurrentHashMap采用分离锁(默认一把全局锁被分离为16把锁，ConcurrentHashMap的实现使用了一个包含16个锁的Array，每一个锁都守护Hash Bucket的1/16；Bucket N由第N mod 16个锁来守护，因此把锁请求减少为原来的1/16且能支持16个并发的写)这种细粒度的锁机制允许更深层次的共享访问，这样可以支持任意数量的线程对ConcurrentHashMap进行读操作，同时支持有限数量的写线程并发修改ConcurrentHashMap，而且读写线程可以并发访问ConcurrentHashMap。

ConcurrentHashMap与其他并发容器所提供的多线程环境下不会抛出并发修改异常的迭代器是由其返回的弱一致性迭代器决定的，弱一致性迭代器可以容许并发修改。当迭代器创建的时，它会遍历已有元素，并且可以感应到在迭代器被创建后对容器的修改。这种弱一致性在调用那些需要对整个容器进行加锁的方法如size或isEmpty时可能提供不精确的值，因此只有当程序需要在独占访问中加锁时，才不能使用ConcurrentHashMap，而在绝大多数情况下ConcurrentHashMap可以带来更好的伸缩性。

ConcurrentHashMap同时将一些常用的复合操作实现为原子操作：

public interface ConcurrentMap<K, V> extends Map<K, V> {
//只有当没有找到匹配K的值时才插入
V putIfAbsent(K key, V value);

//只有当K与V都匹配时才移除
boolean remove(K key, V value);

//只有当K与oldValue都匹配时才替换
boolean replace(K key, V oldValue, V newValue);

//只有当K匹配时才替换
boolean replace(K key, V newValue);
}
在JDK1.6中又加入了ConcurrentSkipListMap来代替同步的SortedMap，加入ConcurrentSkipListSet来代替同步的SortedSet。

(2).写入时复制并发容器：

CopyOnWriteArrayList和CopyOnWriteArraySet这两个写入时复制并发容器用了代替同步的List和Set容器，避免了在迭代期间对容器的加锁和复制，在每次修改时会创建并重新发布一个新的容器拷贝。

当对容器的迭代操作的频率远高于对容器的修改操作的频率时，使用写入时复制容器是个合理的选择。

(3).阻塞队列：

阻塞队列Blocking queue提供了可阻塞的put和take方法，它们与可定时的offer和poll是等价的。如果Queue已满，put方法会被阻塞直到有空间可用；如果Queue已空，take方法会阻塞直到有元素可用。Queue的长度可以有限，也可以无限，无限的Queue永远不会满，因此put方法永远不会被阻塞。阻塞队列非常适合于生产者-消费者模式。

Blocking queue的常见实现有：

A.LinkedBlockingQueue和ArrayBlockingQueue：

提供FIFO队列，比同步List容器拥有更好的性能。

B.PriorityBlockingQueue：

提供按优先级顺序排序的队列。

C.SynchronousQueue：

不是一个真正意义上的队列，不会为队列元素维护任何存储空间，它只维护一个排队的线程清单，这些线程等待把元素加入(enqueue)队列或移出(dequeue)队列。

以洗盘子为例，传统的队列类似于盘架，洗盘子的人(生产者)把洗好的盘子放到盘架上就是入队，烘干盘子的人(消费者)从盘架取走洗好的盘子烘干就是出队；SynchronousQueue就是没有盘架，洗盘子的人直接把洗好的盘子交给烘干盘子的人。

SynchronousQueue实现队列的方式非常直接地移交工作，减少了在生产者和消费者直接移动数据的延迟时间，但是因为它没有存储能力，所以除非另一个线程已经准备好参与移交工作，否则put和take会一直阻塞，因此SynchronousQueue队列只有在消费者充足的情况下比较合适。

(4).双端队列：

JDK1.6新增了两个分别扩展了Queue和BlockingQueue的容器Deque和BlockingDeque，它们是双端队列，允许高效地在队头和队尾进行插入和移除，它们的实现类是ArrayDeque和LinkedBlockingDeque。

双端队列非常适合于工作窃取(work stealing)模式，在生产者-消费者模式中，所有的消费者只能共享一个工作队列，而在工作窃取模式中，每个消费者都有一个自己的双端队列，大多数情况下只访问自己的双端队列，如果一个消费者完成了自己双端队列中的全部工作，它可以偷取其他消费者的双端队列中的末尾任务(注意是从尾部而不是头部获取工作，降低双端队列的竞争)，因为工作线程并不会竞争一个共享的任务队列，所以工作窃取模式比传统的生产者-消费者模式具有更佳的伸缩性。

双端队列的工作窃取模式非常适合解决生产者-消费者模式中生产者也同时是消费者，消费者同时又是生产者的情况。

4.用阻塞队列实现生产者消费者模式：

生产者-消费者模式是一个经典的线程同步问题，可以将生产者和消费者解耦，实现高并发，很多JDK的类库就使用了该模式，比如java并发线程池。传统使用互斥或信号量实现的生产者-消费者模式例子很多，这里以桌面搜索程序的扫描文件和建立索引为例，使用阻塞队列来实现生产者-消费者模式，代码如下：

//扫描文件，生产者
public class FileCrawler implements Runnable{
private final BlockingQueue<File> fileQueue;
private final FileFilter filter;
private final File root;

public FileCrawler(BlockingQueue<File> queue, FileFilter filter, File root){
this.fileQueue = queue;
this.filter = filter;
this.root = root;
}

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

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

//建立索引，消费者
Public class Indexer implements Runnable{
private final BlockingQueue<File> queue;

public Indexer(BlockingQueue<File> queue){
this.queue = queue;
}
......
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 file : roots){
new Thread(new FileCrawler(queue, filter, root)).start();
}
for(int i = 0; i < N_CONSUMERS; i++){
new Thread(new Indexer(queue)).start();
}
}
5.闭锁：

闭锁(Latch)可以延迟线程的进度直到线程达到终止状态，闭锁可以用来确保特定活动直到其他活动完成后才发生。

CountDownLatch是一个灵活的闭锁实现，允许一个或多个线程等待一个事件集的发生，它初始化为一个正整数计数器，用来表示需要等待的事件数，countDown操作对计数器进行减一操作，表示一个事件已经发生，await方法等待计数器到达零，若到达零则表示所有等待的事件都已经发生。

我们以统计线程执行时间为例演示闭锁的使用，代码如下：

public class TestHarness {
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.await();
try{
task.run();
}finally{
endGate.countDown();
}
}catch(InterruptedException ignored){
}
}
};
t.start();
}
long start = System.nanoTime();
startGate.countDown();
endGate.await();
long end = System.nanoTime();
return end - start;
}
}
一个闭锁工作起来就像一扇大门，直到闭锁达到终点状态之前，门一直是关闭的，没有线程能够通过，在终点状态到来的时候，门打开了，允许所有的线程都通过，一旦闭锁达到了终点状态，就再也不能改变状态了，即闭锁是一次性对象，一旦进入最终状态就不能被重置了，这是闭锁和关卡的最大区别。

6.关卡：

关卡(Barrier)类似于闭锁，它们都能阻塞一组线程，直到某些事件发生，其中关卡与闭锁不同之处在于：

A.所有线程必须同时达到关卡点，才能继续出现，闭锁等待的是事件，关卡等待的是其他线程。

B.闭锁是一次性使用的，一旦进入到最终状态，就不能被重置了；关卡可以重复使用。

JDK中常见的关卡有以下两种：

(1).CyclicBarrier：

CyclicBarrier允许一个给定数量的成员多次集中在一个关卡点，当线程达到关卡点时，调用await方法，await会被阻塞直到所有线程都达到关卡点，当所有线程都达到了关卡点，关卡就被成功地突破了，所有线程都被释放，关卡会重置以备下一次使用。若对await调用超时，或者阻塞中的线程被中断，那么关卡就被认为是失败的，所有对await未完成的调用都通过BrokenBarrierException终止。

CyclicBarrier允许向构造方法传递一个Runnable的关卡行为，在阻塞线程被释放之前是不能执行的，当成功通过关卡时，会在一个子任务线程中执行该关卡行为。

在这并行迭代算法中非常有用，这个算法会把一个问题拆分成一些列相互独立的子问题，使用CyclicBarrier关卡的例子代码如下：

public class Solver {
final int N;
final float[][] data;
final CyclicBarrier barrier;

class Worker implements Runnable {
int myRow;

public Worker(int row) {
myRow = row;
}

public void run() {
while (!done()) {
processRow(myRow);
try {
barrier.await();
} catch (InterruptedException ex) {
return;
} catch (BrokenBarrierException ex) {
return;
}
}
}
}

public Solver(float[][] matrix) {
data = matrix;
N = matrix.length;
barrier = new CyclicBarrier(N, new Runnable() {
public void run() {
mergeRows(...);
}
});
for (int i = 0; i < N; ++i){
new Thread(new Worker(i)).start();
}
waitUntilDone();
}
}

(2).Exchanger：

Exchanger是一种两步关卡，在关卡点会交换数据。

Exchanger为线程交换信息提供了非常方便的途径，它可以作为两个线程交换对象的同步点，当一个线程调用Exchange对象的exchange方法后，它会陷入阻塞状态，直到另一个线程也调用了exchange方法，然后以线程安全的方式交换数据，之后这两个线程继续运行，使用Exchanger关卡的例子代码如下：

public class FillAndEmpty {
Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();
DataBuffer initialEmptyBuffer = ......;
DataBuffer initialFullBuffer = ......;

class FillingLoop implements Runnable {
public void run() {
DataBuffer currentBuffer = initialEmptyBuffer;
try {
while (currentBuffer != null) {
addToBuffer(currentBuffer);
if (currentBuffer.isFull()){
currentBuffer = exchanger.exchange(currentBuffer);
}
}
} catch (InterruptedException ex) { ... handle ... }
}
}

class EmptyingLoop implements Runnable {
public void run() {
DataBuffer currentBuffer = initialFullBuffer;
try {
while (currentBuffer != null) {
takeFromBuffer(currentBuffer);
if (currentBuffer.isEmpty()){
currentBuffer = exchanger.exchange(currentBuffer);
}
}
} catch (InterruptedException ex) { ... handle ...}
}
}

public void start() {
new Thread(new FillingLoop()).start();
new Thread(new EmptyingLoop()).start();
}
}
7.FutureTask：

多线程的Runnable任务是不返回执行结果的，Callable任务是携带执行结果的，FutureTask用于获取Callable任务的执行结果，它有3个状态：等待、运行和完成，完成包括所有计算以任意的方式结束，包括正常结束(FutureTask可以取消正在执行的任务)、取消和异常。

FutureTask通过Future的get方法获取任务执行结果，它依赖于任务的执行状态，如果已经完成，get可以立刻得到返回结果，否则会被阻塞直到任务装入运行状态，然后返回结果或抛出异常。FutureTask把执行结果从运行的线程传送到需要这个结果的线程。

Executor线程池框架利用FutureTask来完成异步任务，并可以用来进行任何潜在的耗时计算，而且可以在真正需要结果之前就启动他们开始计算。FutureTask的例子代码如下：

public class Preloader {
private final FutureTask<ProductInfo> future = new FutureTask<ProductInfo>(
new Callable<ProductInfo>() {
public ProductInfo call() throws DataLoadException {
return loadProductInfo();
}
}
);

private final Thread thread = new Thread(future);

public void start() {
thread.start();
}

public ProductInfo get() throws DataLoadException, InterruptedException{
try{
return future.get();
}catch(ExecutionException e){
......
}
}
}
8.信号量：

计数信号量(Semaphore)用来控制能够同时访问某特定资源的活动数量，或者同时执行某一个给定操作的数量。一个信号量管理一个有效的许可集，许可的初始量通过构造函数传入，若还有剩余许可，活动能够获得许可，并在使用之后通过release方法释放许可；若已经没有许可，那么acquire方法会被阻塞，直到有可用许可、中断或超时为止。

计数信号量的一种退化形式是二元信号量：一个计数初始值为1的Semaphore。二元信号量可用做互斥锁，它有不可重入锁的语意。

计数信号量可以用来实现资源池或者给容器定边界，一个使用信号量的例子代码如下：

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 t) throws InterruptedException{
sem.acquire();
boolean wasAdded = false;
try{
wasAdded = set.add(t);
teturn wasAdded;
}finally{
if(!wasAdded){
sem.release();
}
}
}

public boolean remove(T t){
boolean wasRemoved = set.remove(T);
if(wasRemoved){
sem.release();
}
return wasRemoved;
}
}