第五章：基础构建模块——java并发编程实战

委托是创建线程安全类的一个最有效的策略：只需让现有的线程安全类管理所有的状态即可。

一、同步容器类

1、同步容器类的问题

　　同步容器类都是线程安全的，容器本身内置的复合操作能够保证原子性，但是当在其上进行客户端复合操作则需要额外加锁保护其安全性

　　由于同步容器类要遵守同步策略，即支持客户端加锁，但必须清楚加入同一个锁

2、迭代器与ConcurrentModificationException

　　及时失败机制：容器在迭代过程中被修改时 ，就会抛出一个ConcurrentModificationException异常

　　解决方法：加锁或创建副本

3、隐藏迭代器

　　一些隐藏的迭代操作：hashCode, equals, containsAll, removeAll, retainAll等

二、并发容器

同步容器对容器状态访问实现串行化以保证线程安全，但这种方法严重降低了并发性。通过并发容器来代替同步容器，可以极大地提高伸缩性并降低风险。

BlockingQueue扩展了Queue，实现了可阻塞的插入和获取等操作ConcurrentHashMap代替HashMap

1、ConcurrentHashMap

　　并不是在每个方法上都在锁使得只有一个线程可以访问容器，即没有实现独占访问。而是使用一种粒度更细的加锁机制来实现大程度的共享，这种机制称为分段锁(Lock Striping)

　　ConcurrentHashMap的迭代器不会抛出ConcurrentModificationException，因此不需要在迭代过程中加锁，因为其返回的迭代器具有弱一致性，而非"及时失败"。

　　ConcurrentHashMap对一些操作进行了弱化，如size(计算的是近似值，而不是精确值)， isEmpty等

2、额外的原子Map操作

　　ConcurrentHashMap实现了若没有则添加、若有则删除、映射则替换等操作的接口

3、CopyOnWriteArrayList

　　特点：写入时复制，即每当修改容器时都会复制底层数组产生开销，只要发布一个事实不可变的对象，那么在访问该对象时就不需要进一步同步

　　不会抛出ConcurrentModificationException，不用加锁，性能更好

　　仅当迭代操作远远多于修改操作时，才应该使用"写入时复制"容器

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

阻塞队列提供了可阻塞的put和take方法，以及支持定时的offer和poll方法。（offer方法如果数据不能添加到队列则返回一个失败状态）

可有界也可无界

在构建高可靠的应用程序时，有界队列是一种强大的资源管理工具；它们能够意志或防止产生过多的工作项，使应用程序在负荷过载的情况下变得更加健壮。

实现：LinkedBlockingQueue， ArrayBlockingQueue， PriorityBlockingQueue（可实现comparable方法比较排序），SynchronousQueue（维护一组工作线程，而不是维护队列元素的存储空间）

1、串行线程封闭

　　对于可变对象，生产者--消费者这种设计与阻塞队列一起，促进了串行线程封闭，从而将对象所有权从生产者交付给消费者。

　　线程封闭对象由单个线程所有，但通过生产者消费者模式安全的转移了对象的所有权，转移后只有接受的线程获得该对象的所有权，而发布者放弃了所有权，并不会在访问他。

2、双端队列适用于工作密取

　　Deque和BlockingDeque对Queue进行了拓展，实现了双端队列，即从头尾皆可插入删除。（实现：ArrayDeque，LinkedBlockingDeque）

　　每个消费者有各自的双端队列，当消费者自己的双端队列为空时，它会从其他消费者队列末尾中密取任务。优点：大大减少了竞争，保证线程均出于忙碌状态

四、阻塞方法和中断方法

阻塞的原因：等待I/O操作结束，等待获得一个锁，等待从Thread.sleep方法中醒来，或是等待另一个线程的计算结果等

线程阻塞时会被挂起，处于某种阻塞状态（BLOCKED，WAITING，TIMED_WAITING），并且必须等待某个不受他控制的事件完成

抛出InterruptedException的方法叫做阻塞方法

中断是一种协作机制，一个线程不能强制要求其他线程停止正在执行的操作而去执行其他操作。

处理对中断的响应：传递InterreuptedException，抛出异常给方法调用者，或捕获异常，做一些清理工作再抛出抛出异常；恢复中断：有时不能抛出InterruptedException, 比如在Runnable中，则可以恢复中断

五、同步工具类

1、闭锁：确保某些活动直到其他活动都完成后才继续执行

　　闭锁的作用相当于一扇门：在闭锁到达结束状态之前，这扇门一直是关闭的，并且没有任何线程能通过，当到达结束状态时，这扇门会打开并允许所有的线程通过。当闭锁到达结束状态后，将不会再改变状态。

　　CountDownLatch：一个或多个线程等待一组事件发生。闭锁状态包括一个计数器，该计数器被初始化为一个正数，表示需要等待的事件数量。countDown方法递减计数器，表示有一个事件已经发生了，而await方法阻塞直到计数器到达零

public long timeTasks(int nThreads, final Runnable task) throws InterruptedException{
        final CountDownLatch startGate = new CountDownLatch(1); //所有线程同时开始执行task的阀门
        final CountDownLatch endGate = new CountDownLatch(nThreads); //所有线程结束的阀门
        
        for (int i=0; i<nThreads; i++){
            Thread t = new Thread(){
                @Override
                public void run() {
                    try {
                        startGate.await(); //等待startGate值减为0
                        try {
                            task.run();
                        } finally{
                            endGate.countDown(); //一个线程运行结束，值减1
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            t.start();
        }
        long start = System.nanoTime();
        startGate.countDown(); //所有线程开始执行task
        endGate.await(); //等待所有线程执行结束
        long end = System.nanoTime();
        return end - start;
    }

2、FutureTask

　　FutureTask是通过 Callable 来实现的，相当于一种可生成结果的 Runnable，并且可处于以下三种状态：等待运行，正在运行，运行完成（正常完成、取消、异常结束）。当FutureTask进入完成状态后，它会停留在这个状态上。

　　Future.get 用来获取计算结果，如果FutureTask还未运行完成，则会阻塞。FutureTask 将计算结果从执行计算的线程传递到获取这个结果的线程，而FutureTask 的规范确保了这种传递过程能实现结果的安全发布。

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class Preloader {
    private final FutureTask<Integer> future = new FutureTask<>(new Callable() {
        public Integer call() throws Exception {
            return 969*99*99;
        }
    });

    private final Thread thread = new Thread(future);

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

    public Integer get() throws Exception {
        try {
            return (Integer) future.get();
        } catch (ExecutionException e) {
            Throwable cause = e.getCause();
            throw launderThrowable(cause);
        }
    }

    private static Exception launderThrowable(Throwable cause) {
        if (cause instanceof RuntimeException)
            return (RuntimeException) cause;
        else if (cause instanceof Error)
            throw (Error) cause;
        else
            throw new IllegalStateException("Not Checked", cause);
    }

    public static void main(String[] args) throws Exception {
        Preloader p = new Preloader();
        p.start();
        long start = System.currentTimeMillis();
        System.out.println(p.get());
        System.out.println(System.currentTimeMillis() - start);
    }
}

 3、信号量

计数信号量用来控制同时访问某个特定资源的操作数量，或同时执行某个指定操作的数量。或者可以用来实现某种资源池，或者对容器施加边界。

Semaphore管理一组虚拟许可，有构造函数指定数量（数量为1即为互斥锁），acquire请求许可（阻塞直到获得，或中断，或超时），release释放一个许可

使用Semaphore实现有界阻塞容器

public class BoundedList<T> {  
  
    private final List<T> list;  
    private final Semaphore semaphore;  
  
    public BoundedList(int bound) {  
        list = Collections.synchronizedList(new LinkedList<T>());  
        semaphore = new Semaphore(bound);  
    }  
  
    public boolean add(T obj) throws InterruptedException {  
        semaphore.acquire();  
        boolean addedFlag = false;  
        try {  
            addedFlag = list.add(obj);  
        }  
        finally {  
            if (!addedFlag) {  
                semaphore.release();  
            }  
        }  
        return addedFlag;  
    }  
  
    public boolean remove(Object obj) {  
        boolean removedFlag = list.remove(obj);  
        if (removedFlag) {  
            semaphore.release();  
        }  
        return removedFlag;  
    }
}

4、栅栏

栅栏类似于闭锁，它能阻塞一组线程直到某个事件发生

闭锁用于等待事件，而栅栏用于等待其他线程。闭锁是一次性对象，一旦进入终止状态，就不能被重置

六、构建高效可伸缩的结果缓存

首先考虑采用HashMap，通过sychronized方法满足原子性

–> 性能较差，同一时间只有一个线程进行计算操作，使用ConcurrentHashMap改善性能，无需使用同步方法，但是可能导致很多线程在计算同样的值

–> 考虑阻塞方法，使用基于FutureTask的ConcurrentHashMap，Future.get实现阻塞知道结果返回，减少了多次计算，但仍然不是原子性的

–> 使用ConcurrentHashMap中的 putifAbsent()

–> 继续解决缓存污染问题，当缓存结果失效时移除，解决缓存逾期，缓存清理等等问题

public class Memoizer <A, V> implements Computable<A, V> {
    private final ConcurrentMap<A, Future<V>> cache
            = new ConcurrentHashMap<A, Future<V>>();
    private final Computable<A, V> c;
 
    public Memoizer(Computable<A, V> c) {
        this.c = c;
    }
 
    public V compute(final A arg) throws InterruptedException {
        while (true) {
            Future<V> f = cache.get(arg);
            if (f == null) {
                Callable<V> eval = new Callable<V>() {
                    public V call() throws InterruptedException {
                        return c.compute(arg);
                    }
                };
                FutureTask<V> ft = new FutureTask<V>(eval);
                f = cache.putIfAbsent(arg, ft);
                if (f == null) {
                    f = ft;
                    ft.run();
                }
            }
            try {
                return f.get();
            } catch (CancellationException e) {
                cache.remove(arg, f);
            } catch (ExecutionException e) {
                throw LaunderThrowable.launderThrowable(e.getCause());
            }
        }
    }
}