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

末蓝、 2022-03-25 13:20 250阅读 0赞

### 《java并发编程实战笔记》 ###

*  第五章 基础构建模块
 *   *  同步容器类在并发下存在的问题
     *  容器类的迭代器并发面临的问题
     *  并发容器
     *   *  ConcurrentHashMap
         *  CopyOnWriteArrayList
         *  阻塞队列BlockingQueue和生产者---消费者模式
     *  阻塞方法与中断方法
     *  同步工具类
     *   *  闭锁
         *  Future Task
         *  信号量
         *  栅栏
     *  构建高效且可伸缩的结果缓存
     *   *  使用HashMap和同步机制来初始化缓存
         *  用ConcurrentHashMap替换HashMap
         *  使用FutureTask实现真正意义上缓存

# 第五章 基础构建模块 #

经典的同步容器类有Vector和Hashtable，两者是早期JDK出现的，此外还有一些由Collections.synchronizedXxx等工厂创建的同步封装器类。这些同步容器类实现的方式基本上都是将它们的状态封装起来，并对每个公有方法都进行同步，使每次只有一个线程能访问容器的状态。

## 同步容器类在并发下存在的问题 ##

首先容器中常见的复合操作包括：迭代、跳转(根据指定顺序找到当前原始的下一个元素)、以及类似“若没有则添加”的条件运算。在同步容器类中，这些复合操作在单线程操作下仍然是安全的，单并发修改容器时，它们可能会出现出乎意料的意外。**因此，同步容器类可能需要额外的客户端加锁来保护复合操作**。常见的、经典的、并发下同步容器类可能会出现的问题有：  
  例如：Vector中有getLast和deleteLast方法，它们都会执行“先检查再运行”操作，即首先获得Vector的大小，在通过结果获取或者删除最后一个元素。在多线程并发情况下，试想A 、B线程同时，同时通过getLast、deleteLast调用一个有10个元素的Vector list，A线程执行getLast方法得到size是10，B线程执行deleteLast先得到的size也是10，刚好B先执行list.remove(9)。执行完后，A线程仍然执行list.get(10);方法，此时会抛出ArrayIndexOutOfBoundsException异常。

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70]  
  利用第四章，对象的组合中客户端加锁方式，通过获得容器类list的锁，可以使getLast、deleteLast方法成为原子操作，并确保Vector的大小在调用size和get之间不会发生变化。

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

再例如：可能抛出ArrayIndexOutOfBoundsException的迭代操作。在调用size、 get的迭代过程中，若有另外一个线程修改了Vector vector

//可能抛出ArrayIndexOutOfBoundsException的迭代操作
    	for (int i = 0; i< vector.size(); i++)
    	{ 
    		doSomething(vector.get(i));
    	}

客户端加锁方式改进：

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

## 容器类的迭代器并发面临的问题 ##

在许多现代的容器类中也有类似Vector中的同步问题，当发现容器在迭代过程中被修改时，就会抛出一个ConcurrentModificationException。无论直接迭代还是使用for-each循环，对容器类进行迭代的标准方式都是使用Iterator，Iterator机制的实现方式是将计数器的变化和容器关联起来，如果迭代期间计数器被修改，那么hasNext或next方法就会抛出异常。 **如果有多个线程并发修改容器，采用迭代器时需要在迭代期间对容器加锁。**  
  但是，长时间地对容器加锁会降低程序的可伸缩性，迭代期持有的锁时间越长，在锁上竞争可能越激烈，迭代器客户端加锁方式性能会受到影响。  
  如果不希望迭代期对容器加锁，还有一种替代方法就是“克隆”容器=，并在副本上进行迭代。副本会被封闭与线程内，因此其他线程不会在迭代期间对副本修改，避免出现抛出ConcurrentModificationException。**此方法关键在于在克隆过程中进行对容器的加锁，但在克隆容器时会存在显著的性能开销**。  
  你必须记住所有对共享容器进行迭代的地方都需要加锁，实际情况会更加复杂，\*\*因为某些情况下，迭代器会隐藏起来。\*\*书中提供了HiddenIterator 例子，在调用System.out.println("debug: added ten elements to "+set); 过程中会执行对set的toString方法，标准容器的toString方法会隐式迭代容器。TestHiddenIteratorThread 是自己编的进行测试，Thread1在执行System.out.println("debug: added ten elements to "+set); 隐式调用了set的迭代器，但是Thread2可能又同时更改了set，因此会出现ConcurrentModificationException。

public class HiddenIterator { 
    	private final Set<Integer> set = new HashSet<>();
    	public synchronized void add(Integer i) { set.add(i);}
    	public synchronized void remove(Integer i) { set.remove(i);}
    	public void addTenThings()
    	{ 
    		Random random = new Random();
    		for(int i = 0; i<10; i++)
    		{ 
    			add(random.nextInt());
    			System.out.println("debug: added ten elements to "+set);	
    		}
    	}
    }
    public class TestHiddenIteratorThread implements Runnable{ 
    	HiddenIterator hiddenIterator = new HiddenIterator();
    	@Override
    	public void run() { 
    		System.out.println(Thread.currentThread().getName());
    		hiddenIterator.addTenThings();
    	}
    	public static void main(String[] args) { 
    		TestHiddenIteratorThread testHiddenIteratorThread = new TestHiddenIteratorThread();
    		new Thread(testHiddenIteratorThread, "Thread1").start();
    		new Thread(testHiddenIteratorThread, "Thread2").start();
    	}
    }

测试出现的异常：  
![在这里插入图片描述][20190121163827450.png]  
  当然，原因在于HiddenIterator不是线程安全的，在执行addTenThings前应该先获取HiddenIterator的锁。但是，如果你没有意识到System.out.println(set);会隐式执行set的迭代器，你会纳闷多个线程同时执行synchronized add();方法为何会报错。当然，此例子中的set改成Collections.synchronizedSet(new HashSet<>())也行。

## 并发容器 ##

虽然已经有了同步容器类(Vector、Stack、HashTable和Collections类中提供的静态工厂方法创建的类)，但是同步容器类基本上的方法都采用了synchronized进行了同步，很明显必然会影响到执行性能。于是乎，java5.0提供了多种并发容器类，例如ConcurrentHashMap、CopyOnWriteArrayList。

### ConcurrentHashMap ###

HashMap线程不安全，Hashtable线程安全但是并发操作独占一把锁，性能不行。因此，新增了强大的ConcurrentHashMap并发容器类，ConcurrentHashMap并不是将每个方法都用在同一个锁上同步使每次一个线程访问容器，而是**采用一种粒度更细的加锁机制(分段锁)来实现更大程度的共享**，任意数量的读写线程可以并发访问Map，并且ConcurrentHashMap提供的迭代器**具有弱一致性**，不会抛出ConcurrentModificationException，**迭代过程不需要进行加锁**。  
  相比Hashtable和synchronizedMap，ConcurrentHashMap拥有更多的优势。ConcurrentHashMap比synchronizedMap有着更好的伸缩性(扩展性)。只有当应用程序需要加锁Map进行独占访问，才应该放弃使用ConcurrentHashMap。同时由于ConcurrentHashMap不能加锁来执行独占访问，我们也无法使用客户端来创建新的原子复合操作，但是ConcurrentHashMap也比较人性，一般基本复合操作例如“若没有则添加、若相等则移除”都在ConcurrentMap接口中声明，而ConcurrentHashMap刚好实现了该接口。

### CopyOnWriteArrayList ###

CopyOnWriteArrayList用于替代同步List(很神奇我没有找到SynchronizedList)，这个容器的线程安全性在于，只要正确地发布一个事实不可变的对象，那么在访问该对象时就不再需要进一步同步。这个容器不会抛出ConcurrentModificationException，并且返回的元素与迭代器创建时元素完全一致，而不必考虑之后修改带的影响。

### 阻塞队列BlockingQueue和生产者—消费者模式 ###

**阻塞队列(BlockingQueue)**：阻塞队列提供了可阻塞的put和take方法，以及支持定时的offer和poll方法，如果队列已经满了，那么put方法将阻塞直到有可用空间。如果队列为空，那么take方法将会阻塞直到有元素可用。队列可以是有界的也可以是无界的。阻塞队列会对当前线程产生阻塞，比如一个线程从一个空的阻塞队列中取元素，此时线程会被阻塞直到阻塞队列中有了元素。当队列中有元素后，被阻塞的线程会自动被唤醒（不需要编写代码去唤醒）。在并发编程中，一般推荐使用阻塞队列，这样实现可以尽量地避免程序出现意外的错误。  
  **生产者----消费者模式**：生产者----消费者模式有着以下几点优点：  
a、解耦，假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法，那 么生产者对于消费者就会产生依赖（也就是耦合）。将来如果消费者的代码发生变化， 可能会影响到生产者。而如果两者都依赖于某个缓冲区，两者之间不直接依赖，耦合也 就相应降低了。  
b、支持并发，由于生产者与消费者是两个独立的并发体，他们之间是用缓冲区作为桥梁连接，生产者只需要往缓冲区里丢数据，就可以继续生产下一个数据，而消费者只需要从缓冲区了拿数据即可，这样就不会因为彼此的处理速度而发生阻塞。  
  **常用的队列**主要有以下两种：（当然通过不同的实现方式，还可以延伸出很多不同类型的队列，DelayQueue就是其中的一种）  
　　先进先出（FIFO）：先插入的队列的元素也最先出队列，类似于排队的功能。从某种程度上来说这种队列也体现了一种公平性。  
　　后进先出（LIFO）：后插入队列的元素最先出队列，这种队列优先处理最近发生的事件  
   **BlockingQueue不同于一般的队列，同时简化了生产者----消费者设计的实现过程**。**它支持任意数量的生产者和消费者**。利用BlockingQueue，当我们有若干生产者线程，另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，及时两者处理数据的速率不同，仍然可以很方便地解决他们之间的数据共享问题。一种常见的生产者-消费者模式就是线程池与工作队列的组合。在Executor任务执行框架中就体现了这种模式。  
   **BlockingQueue的核心方法**：  
　　1.放入数据  
　　　　a、offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.（本方法不阻塞当前执行方法 的线程）；　　　　　　  
　　　　b、offer(E o, long timeout, TimeUnit unit)：可以设定等待的时间，如果在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。  
　　　　c、put(an Object):把an Object加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续。  
　　2. 获取数据  
　　　　 a、poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null;  
　　　　 b、poll(long timeout, TimeUnit unit)：从BlockingQueue取出一个队首的对象，如果在指定时间内，队列一旦有数据可取，则立即返回队列中的数据。否则知道时间超时还没有数据可取，返回失败。  
　　　　c、take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入;  
　　　　d、drainTo():一次性从BlockingQueue获取所有可用的数据对象（还可以指定获取数据的个数），通过该方法，可以提升获取数据效率，不需要多次分批加锁或释放锁。  
 **常见BlockingQueue**：有待进一步详细学习  
**LinkedBlockingQueue 和ArrayBlockingQueue是FIFO队列，两者分别与LinkedList和ArrayList相似，但比同步List有更好的并发性能。**  
　1. **ArrayBlockingQueue**：基于数组的阻塞队列实现，在ArrayBlockingQueue内部，维护了一个定长数组，以便缓存队列中的数据对象，这是一个常用的阻塞队列，除了一个定长数组外，ArrayBlockingQueue内部还保存着两个整形变量，分别标识着队列的头部和尾部在数组中的位置。ArrayBlockingQueue在生产者放入数据和消费者获取数据，都是共用同一个锁对象，由此也意味着两者无法真正并行运行，这点尤其不同于LinkedBlockingQueue；按照实现原理来分析，ArrayBlockingQueue完全可以采用分离锁，从而实现生产者和消费者操作的完全并行运行。Doug Lea这个人之所以没这样去做，也许是因为ArrayBlockingQueue的数据写入和获取操作已经足够轻巧，以至于引入独立的锁机制，除了给代码带来额外的复杂性外，其在性能上完全占不到任何便宜。  
　 　 ArrayBlockingQueue和LinkedBlockingQueue间还有一个明显的不同之处在于，前者在插入或删除元素时不会产生或销毁任何额外的对象实例，而后者则会生成一个额外的Node对象。这在长时间内需要高效并发地处理大批量数据的系统中，其对于GC的影响还是存在一定的区别。而在创建ArrayBlockingQueue时，我们还可以控制对象的内部锁是否采用公平锁，默认采用非公平锁。  
　　2.**LinkedBlockingQueue** ：基于链表的阻塞队列，同ArrayListBlockingQueue类似，其内部也维持着一个数据缓冲队列（该队列由一个链表构成），当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；只有当队列缓冲区达到最大值缓存容量时（LinkedBlockingQueue可以通过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反之对于消费者这端的处理也基于同样的原理。而LinkedBlockingQueue之所以能够高效的处理并发数据，还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。  
　　3. **DelayQueue**：DelayQueue中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQueue是一个没有大小限制的队列，因此往队列中插入数据的操作（生产者）永远不会被阻塞，而只有获取数据的操作（消费者）才会被阻塞。 　　  
　　使用场景：DelayQueue使用场景较少，但都相当巧妙，常见的例子比如使用一个DelayQueue来管理一个超时未响应的连接队列。  
　　4. **PriorityBlockingQueue**： 基于优先级的阻塞队列（优先级的判断通过构造函数传入的Compator对象来决定），但需要注意的是PriorityBlockingQueue并不会阻塞数据生产者，而只会在没有可消费的数据时，阻塞数据的消费者。因此使用的时候要特别注意，**生产者生产数据的速度绝对不能快于消费者消费数据的速度，否则时间一长，会最终耗尽所有的可用堆内存空间**。在实现PriorityBlockingQueue时，内部控制线程同步的锁采用的是公平锁。  
　　5. **SynchronousQueue ：一种无缓冲的等待队列，因此take和put会一直阻塞**，类似于无中介的直接交易，有点像原始社会中的生产者和消费者，生产者拿着产品去集市销售给产品的最终消费者，而消费者必须亲自去集市找到所要商品的直接生产者，如果一方没有找到合适的目标，那么对不起，大家都在集市等待。相对于有缓冲的BlockingQueue来说，少了一个中间经销商的环节（缓冲区），如果有经销商，生产者直接把产品批发给经销商，而无需在意经销商最终会将这些产品卖给那些消费者，由于经销商可以库存一部分商品，因此相对于直接交易模式，总体来说采用中间经销商的模式会吞吐量高一些（可以批量买卖）；但另一方面，又因为经销商的引入，使得产品从生产者到消费者中间增加了额外的交易环节，单个产品的及时响应性能可能会降低。声明一个SynchronousQueue有两种不同的方式，它们之间有着不太一样的行为。  
　　公平模式和非公平模式的区别: 　　  
　　如果采用公平模式：SynchronousQueue会采用公平锁，并配合一个FIFO队列来阻塞多余的生产者和消费者，从而体系整体的公平策略； 　　  
　　但如果是非公平模式（SynchronousQueue默认）：SynchronousQueue采用非公平锁，同时配合一个LIFO队列来管理多余的生产者和消费者，而后一种模式，如果生产者和消费者的处理速度有差距，则很容易出现饥渴的情况，即可能有某些生产者或者是消费者的数据永远都得不到处理。

## 阻塞方法与中断方法 ##

线程阻塞与执行时间很长的普通操作差别在于：被阻塞的线程必须等待某个不受它控制的事件发生后才能继续执行，例如等待某个锁编程可用。阻塞队列的put和take方法会抛出异常Interrupted Exception，当某个方法抛出Interrupted Exception时，表明该方法是一个阻塞方法。  
  中断是一种协作机制，一个线程不能强制要求其他线程停止正在执行的操作而去执行其他操作。当线程A中断线程B时，A只是要求B在执行到某个可以暂停的地方停止正在执行的操作。但是实际怎样处理中断是由线程B自己决定的。所以在，类中调用阻塞方法时，需要添加中断处理。  
  **常见会用到中断情形**：  
    a、点击某个桌面应用中的取消按钮时；  
    b、某个操作超过了一定的执行时间限制需要中止时；  
    c、多个线程做相同的事情，只要一个线程成功其它线程都可以取消时；  
    d、一组线程中的一个或多个出现错误导致整组都无法继续时；  
    e、当一个应用或服务需要停止时。  
  **中断的必要条件**：  
  只有阻塞方法才可以中断，因为它提供了中断响应的策略，非堵塞方法强制中断线程不安全。中断后不会产生问题。但是在非中断方法中，比如线程在执行一个排序算法，那么在排序过程中并不会发生中断，因为此时中断会产生问题即排序的数组中一部分有序一部分无序。此时如果在排序排序过程中必须处理才能中断。此时如果要强制结束线程必须调用Thread.stop()或者Thread.destroy()，但是这样强制结束线程是不安全的。 例如下面的例子：当线程中断时，还是会继续输出2和3。当下面的Thread.currentThread().interrupt();改成 Thread.currentThread().stop();或者 Thread.currentThread().destroy();，才只输出结果1。

public class InterruptedExample implements Runnable { 
        @Override
        public void run() { 
            System.out.println("1");
            Thread.currentThread().interrupt();
            System.out.println("2");
            System.out.println("3");
        }
        public static void main(String[] args) { 
            Thread interruptedThread = new Thread(new InterruptedExample());
            interruptedThread.start();
        }
    }

Thread提供了interrupt方法，用于中断线程或者查询线程是否已经被中断。最常用中断的情况就是取消某个操作，当你的代码中调用了一个将抛出InterruptedException异常的方法，你的方法也就变成了阻塞方法，必须采取措施处理对中断的响应，  
**两种处理抛出阻塞方法的方式**：  
  传递：避开这个异常通常是最明智的策略，只需要将InterruptedException传递给方法的调用者，你可以根本不捕获该异常，或者捕获该异常然后执行某种简单的清理工作后再次抛出这个异常。  
  恢复中断：有些interruptedException情况下,例如当代码是Runnable的一部分时，只能必须捕获InterruptedException可通过调用当前线程上的interrupt方法恢复中断状态，这样在调用栈中更高层的代码中将看到引发了一个中断。

public class TaskRunnable implements Runnable{ 
    		BlockingQueue<Task> queue;
    		...
    		public void run(){ 
    			try{ 
    					processTask(queue.take());
    				}
    				catch(InterruptedException e)
    				{ 
    					Thread.currentThread().interrupt();
    				}
    		}
    	}

**中断方法：** public void interrupt():中断线程 ; public static boolean interrupted():查询当前线程中断状态，如果已经中断，就清除中断; public boolean isInterrupted():查询当前线程中断状态，但是不会改变中断状态。  
  当线程正在调用阻塞方法的过程中发生中断，阻塞方法停止并且会抛出InterruptedException异常。如果捕捉到异常但不进行处理，那么线程会继续执行。例如InterruptedExample1 ，在run中捕获了Thread.sleep(10000);阻塞方法的异常，线程仍然执行。

public class InterruptedExample1 implements Runnable { 
        @Override
        public void run() { 
            System.out.println("before interrupt");
            //阻塞10秒
            try { 
                Thread.sleep(10000);
            } catch (InterruptedException e) { 
                System.out.println("interrupted happen");
                //return;
            }
            //捕捉到异常后，线程状态变为没有中断
            System.out.println("isInterrupted:" + Thread.currentThread().isInterrupted());
            System.out.println("after interrupted");
        }
        public static void main(String[] args) throws InterruptedException { 
        	Thread interruptedThread1 = new Thread(new InterruptedExample1(), "Thread1");
        	interruptedThread1.start();
        		 Thread.sleep(2000);      
            System.out.println("after sleep");
            interruptedThread1.interrupt();  
        }
    }

运行结果：  
![在这里插入图片描述][20190122110527680.png]

## 同步工具类 ##

在容器类中，阻塞队列是一种独特的类，它们不仅能作为保存对象的容器，还能协调生产者和消费者等线程之间的控制流，因为take和put等方法将阻塞，直到队列达到期望的状态(队列既非空也非满)。**同步工具类可以是任何一个对象，只要它根据其自身状态来协调线程的控制流**。同步工具类可以是阻塞队列、信号量Semaphore、栅栏Barrier、闭锁。

### 闭锁 ###

闭锁是一个同步工具类，**可以延迟线程的进度直到其到达终止状态**。简单地说，闭锁可以用来确保某些活动直到其它活动都完成后才继续执行。闭锁作用相当于一扇门：当闭锁达到结束状态之前，这扇门一直是关闭的，并且没有任何线程能通过，当到达结束状态时，这扇门会打开并且允许所有的线程通过。当闭锁达到结束状态后，将不会再改变状态，因此此扇门将永远保持打开状态。闭锁可以用来确保某些活动指导其他活动都完成后才继续执行。例如：  
　 　 a、确保某个计算在其需要的所有资源都被初始化之后才继续执行。  
　 　 b、确保某个服务在其依赖的所有其他服务都已经启动之后才启动。  
　 　 c、等待直到某个操作的所有参与者都就绪再继续执行。  
　 CountDownLatch是一种灵活的闭锁实现，可以在上面a 、b、c情形中使用，它可以是一个或者多个线程等待一组时间发生。闭锁状态包括一个计数器，该计数器被初始化为一个正数，表示需要等待的事件数量。countDown方法递减计数器，表示有一个事件发生了，而await方法等待计数器达到零，表示需要等待的事件都已发生，如果计数器的值非零，那么await会一直阻塞直到计数器为0，或者等待中的线程中断、或者等待超时。  
**CountDownLatch的用法**：  
　 典型用法1：**某一线程在开始运行前等待n个线程执行完毕**。将CountDownLatch的计数器初始化为n new CountDownLatch(n) ，每当一个任务线程执行完毕，就将计数器减1 countdownlatch.countDown()，当计数器的值变为0时，在CountDownLatch上 await() 的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。  
　 典型用法2：**实现多个线程开始执行任务的最大并行性**。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。**类似于赛跑**，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的CountDownLatch(1)，将其计数器初始化为1，多个线程在开始执行任务前首先 coundownlatch.await()，当主线程调用 countDown() 时，计数器变为0，多个线程同时被唤醒。  
**CountDownLatch的不足：**  
　 CountDownLatch是一次性的，计数器的值只能在构造方法中初始化一次，之后没有任何机制再次对其设置值，当CountDownLatch使用完毕后，它不能再次被使用。  
　 书上举的例子：通过闭锁CountDownLatch计算多线程下并发执行任务的执行时间，程序中有两个CountDownLatch, startGate为启动门，用于控制所有的线程都启动后才开始同时执行。 endGate为终止门，用于控制所有的线程都执行结束才终止计时。startGate能使主线程同时释放所有的工作线程，而结束门能使主线程等待最后一个线程执行完成，而不是顺序地等待每个线程执行完成。

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();//等待startGate的count为0，否则一直阻塞在这 
    					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;
    	}

### Future Task ###

FutureTask也可以用作闭锁。（FutureTask实现了Future语义，表示一种抽象的可生成结果的计算。FutureTask表示的计算是通过Callable来实现的，相当于一种可生成结果的Runnable，并且可以处于以下3种状态：等待运行（Waiting to run），正在运行（Running）和运行完成（Completed）。”执行完成”表示计算的所有可能结束方式，包括正常结束、由于取消而结束和由于异常而结束等。当FutureTask进入完成状态后，它会停止在这个状态上。  
  FutureTask 有点类似Runnable，都可以通过Thread来启动，不过FutureTask可以返回执行完毕的数据，并且FutureTask的get方法支持阻塞。由于FutureTask可以返回执行完毕的数据，并且FutureTask的get方法支持阻塞这两个特性，我们可以用来预先加载一些可能用到资源，然后要用的时候，调用get方法获取（如果资源加载完，直接返回；否则继续等待其加载完成）。  
书上的例子：

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();
            System.out.println("start");
        }
        public ProductInfo get()
                throws DataLoadException, InterruptedException { 
            try { 
                System.out.println("get start");
                ProductInfo per = future.get();
                System.out.println("get end");
            } catch (ExecutionException e) { 
                Throwable cause = e.getCause();
                if (cause instanceof DataLoadException)
                    throw (DataLoadException) cause;
                else { 
                    throw LaunderThrowable.launderThrowable(cause);
                   // return null;
                }
            }
        }
    }

其他地方看到的例子：

public class PreLoaderUseFutureTask  
    {   
        /** * 创建一个FutureTask用来加载资源 */  
        private final FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() { 
        	 @Override  
             public String call() throws Exception  
             {   
                 Thread.sleep(3000);  
                 return "加载资源需要3秒";  
             }  
    	});
      
        public final Thread thread = new Thread(futureTask);  
      
        public void start()  
        {   
            thread.start();  
        }  
      
        /** * 获取资源 */  
        public String getRes() throws InterruptedException, ExecutionException  
        {   
            return futureTask.get();//加载完毕直接返回，否则等待加载完毕 
        }  
        public static void main(String[] args) throws InterruptedException, ExecutionException  
        {     
            PreLoaderUseFutureTask task = new PreLoaderUseFutureTask();  
            /** * 开启预加载资源 */  
            task.start();  
            // 用户在真正需要加载资源前进行了其他操作了2秒 
            Thread.sleep(2000);    
            /** * 获取资源 */  
            System.out.println(System.currentTimeMillis() + "：开始加载资源");  
            String res = task.getRes();  
            System.out.println(res);  
            System.out.println(System.currentTimeMillis() + "：加载资源结束");  
        }  
    }

### 信号量 ###

计数信号量(Counting Semaphore)用来控制同时访问特定资源的操作数目，或者同时执行某个指定操作的数量。Semaphore中管理者一组虚拟的许可，许可的初始数量可通过构造函数来指定，在执行操作时可以首先获得许可，并在使用之后释放许可，如果没有许可，那么acquire 将阻塞直到有许可（或者被中断或者操作超时）。Semaphore可以用于实现资源池，例如数据库连接池。我们可以构造一个固定长度的资源池，当从池中获取一个资源之前首先调用acquire方法获取一个许可，在资源返回给池之后调用release释放许可，那么acquire将一直阻塞直到资源池不为空。  
  可以使用Semaphore将任何一种容器变成有界阻塞容器，Semaphore就是操作系统中的信号量,是为了解决资源分配问题，限制同一时刻访问某一资源的线程个数。 参考了鸿洋大神的博客。

public class SemaphoreTest { 
        /** * 定义初始值为1的信号量 */  
        private final Semaphore semaphore = new Semaphore(1);  
        /** * 模拟打印操作 * @param str * @throws InterruptedException */  
        public void print(String str) throws InterruptedException  
        {   
            //请求许可 
            semaphore.acquire();  
            //doSomething();
            System.out.println(Thread.currentThread().getName()+" enter ...");  
            Thread.sleep(1000);  
            System.out.println(Thread.currentThread().getName() + "正在打印 ..." + str);  
            System.out.println(Thread.currentThread().getName()+" out ...");  
            //释放许可 
            semaphore.release();  
        }  
      
        public static void main(String[] args)  
        {   
            final SemaphoreTest print = new SemaphoreTest();   
            /** * 开启10个线程，抢占打印机 */  
            for (int i = 0; i < 10; i++)  
            {   
                new Thread()  
                {   
                    public void run()  
                    {   
                        try  
                        {   
                            print.print("helloworld");  
                        } catch (InterruptedException e)  
                        {   
                            e.printStackTrace();  
                        }  
                    };  
                }.start();  
            }  
        }  
    }

### 栅栏 ###

栅栏Barrier类似闭锁，能阻塞**一组线程**直到某个时间发生，与阻塞关键区别在于：**所有的线程必须同时到达栅栏位置，才能继续执行**。如果对await的调用超时，或者await阻塞的线程被中断，那么栅栏就被认为是打破了，所有阻塞的await调用都将终止并抛出BrokenBarrierException，如果成功地通过栅栏，那么await将会为每个线程返回一个唯一的达到索引号，我们可以利用这些索引“选举”产生一个领导线程，并在下一次迭代中由该线程领导之下一些特殊工作。

public class TestcyclicBarrier { 
    	public static void main(String[] args) throws InterruptedException { 
    		CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new Runnable() { 
    	        @Override
    	        public void run() { 
    	            // TODO Auto-generated method stub
    	            System.out.println("三个线程到达栅栏后执行一次");
    	        }
    	    });
    	    for (int i = 0; i < 12; i++) { 
    	        //短暂休眠，使效果明显
    	        Thread.sleep(100);
    	        Thread s = new Thread(new Runnable() { 
    	            @Override
    	            public void run() { 
    	                try { 
    	                    System.out.println(Thread.currentThread().getName() + "Begin");
    	                    cyclicBarrier.await();
    	                    System.out.println(Thread.currentThread().getName() + "End");
    	                } catch (InterruptedException e) { 
    	                    // TODO Auto-generated catch block
    	                    e.printStackTrace();
    	                } catch (BrokenBarrierException e) { 
    	                    // TODO Auto-generated catch block
    	                    e.printStackTrace();
    	                }
    	            }
    	        });
    	        s.start();
    	    }
    	}
    	}

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 1]

## 构建高效且可伸缩的结果缓存 ##

**几乎所有的服务器都会使用某种形式的缓存，重用之前的计算结果能降低延迟、提高吞吐量，但是需要消耗更多的内存，时间换空间的代价**。  
  书中从HashMap和同步机制来初始化缓存，改进至ConcurrentHashMap替换HashMap，再到基于FutureTask的Memorizing封装器，到最后最终实现，自己看着加了测试代码，有不足之处感谢指正。

### 使用HashMap和同步机制来初始化缓存 ###

首先有一个Computable<A,V>接口：

public interface Computable<A,V>{ 
    	V compute(A arg) throws InterruptedException;
    }

其次自己测试的一个耗时长的TimeConsumingComputor 实现了Computable<A,V>接口。

public class TimeConsumingComputor implements Computable<Integer, Integer>{ 
    	@Override
    	public Integer compute(Integer arg) throws InterruptedException { 
    		for(long i = 0; i < 10000000; i++)
    			;
    		return arg;
    	}
    }

然后出现第一代使用HashMap和同步机制当缓存器，HashMap不是线程安全的，因此采用同步机制确保线程安全，但是出现一个明显的可伸缩性问题，每次只能有一个线程执行compute，当compute计算耗时长时，其他线程可能会阻塞很长时间，显然并发性很糟糕。

public class Memoizer1 <A,V> implements Computable<A,V>{ 
    	private final Map<A, V> cache= new HashMap<A,V>();
    	private final Computable<A,V> computor;
    	public Memoizer1(Computable<A, V> c)
    	{ 
    		this.computor = c;
    	}
    	public synchronized V compute(A arg) throws InterruptedException{ 
    		V result = cache.get(arg);
    		if(result == null)
    		{ 
    			result = computor.compute(arg);
    			cache.put(arg, result);
    		}
    		return result;
    	}
    }

下面是测试第一代的缓存器：

public class TestMemorizerThread implements Runnable { 
    	private TimeConsumingComputor timeConsumingComputor = new TimeConsumingComputor();
    	private Memoizer1<Integer, Integer> memorizer1 = new  Memoizer1<Integer, Integer>(timeConsumingComputor);
    	@Override
    	public void run() { 
    		// TODO Auto-generated method stub
    		long startTime = System.currentTimeMillis();
    		for(int i = 0; i < 100; i++)
    		{ 
    			try { 
    				System.out.println(Thread.currentThread().getName()+" input = "+ i +" result = " + memorizer1.compute(i));
    			} catch (InterruptedException e) { 
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    		long endTime = System.currentTimeMillis();
    		System.out.println(Thread.currentThread().getName()+" cost time is "+String.valueOf(endTime - startTime));
    	}
    	public static void main(String[] args) throws InterruptedException { 
    		TestMemorizerThread testMemorizerThread = new TestMemorizerThread();
    		new Thread(testMemorizerThread, "thread1").start();
    	// Thread.sleep(2000); 
    		new Thread(testMemorizerThread, "thread2").start();
    		new Thread(testMemorizerThread, "thread3").start();	
    	}
    }

首先可以肯定是缓存器是肯定有效的，如果让thread1先执行，Thread.sleep(2000)后，再执行thread2、thread3，可以看出执行的速度  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 3]  
  但是实际并发执行时，线程1 2 3都会去计算相同的i，三个线程计算相同的东西，并且是排队阻塞计算，都要写入缓存，缓存的作用是避免相同结果计算多次，显然第一代缓存器很糟糕。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 4]

### 用ConcurrentHashMap替换HashMap ###

第二代缓存器，用ConcurrentHashMap替换了HashMap和同步机制

public class Memoizer2<A,V> implements Computable<A, V> { 
    	private final Map<A,V> cache = new ConcurrentHashMap<A,V>();
    	private final Computable<A, V> computor ;
    	public Memoizer2(Computable<A, V> c) { 
    		this.computor = c;
    	}
    	@Override
    	public V compute(A arg) throws InterruptedException { 
    		V result = cache.get(arg);
    		if(result == null)
    		{ 
    			result =  computor.compute(arg);
    			cache.put(arg, result);
    		}
    		return result;
    	}
    }

显然运行结果并没有快很多，但是第一代的三个线程总执行时间为741\*3，第二代由于是并行计算，因此总执行时间是862。但是，同一个值还是三个线程都会去计算，并没有真正实现缓存的意义。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 5]

### 使用FutureTask实现真正意义上缓存 ###

**真正理想的状态是，线程1正在计算f(1)，线程2 3 也打算计算f(1)，但他们知道最高效的方式是等线程1计算完后，再去查询缓存。而Memoizer2是多个线程并发判断是否f(1)结果存在，有则直接取结果。更加严格意义上的缓存必然是，线程判断f(1)是否有其他线程开始计算，如果还没有线程启动，那么就创建一个FutureTask并注册到Map中，然后启动计算。** 通过构建Callable和Future的线程，不同于实现Runnable和继承Thread方式，在任务执行完成后就可以获取执行结果。Callable类似Runnabel接口，都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor执行，ThreadPoolExecutor或ScheduledThreadPoolExecutor都实现了ExcutorService接口，而因此Callable需要和Executor框架中的ExcutorService结合使用，ExecutorService提供了submit提交一个实现Callable接口的任务，并且返回封装了异步计算结果的Future。FutureTask类除了实现了Future接口外还实现了Runnable接口，因此FutureTask也可以直接提交给Executor执行。 当然也可以调用线程直接执行（FutureTask.run()）。  
最终可以通过Future对象.get() 判断异步计算的结果。

public class Memoizer3<A,V> implements Computable<A, V> { 
    	private final Map<A, Future<V>> cache = new ConcurrentHashMap<>();
    	private final Computable<A, V> computor;
    	public Memoizer3(Computable<A, V> c) { this.computor = c;}
    	public V compute(final A arg) throws InterruptedException{ 
    		Future<V> future = cache.get(arg);
    		if(future == null)
    		{ 
    			Callable<V> eval  = new Callable<V>() { 
    				public V call() throws InterruptedException{ 
    					return computor.compute(arg);
    				}
    			};
    			FutureTask<V> fTask = new FutureTask<>(eval);
    			future  = fTask;
    			cache.put(arg, fTask);
    			fTask.run();//调用一个新线程直接执行（FutureTask.run()）
    		}
    		try { 
    			return future.get(); //返回Callable中call方法结果
    		} catch (Exception e) { 
    			e.printStackTrace();
    			throw LaunderThrowable.launderThrowable(e.getCause());
    		}
    	}
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 6]  
  Memoizer表现出非常好的并发性（基本上是因为发挥了ConcurrentHashMap高效的并发性），若计算结果计算出那么立即返回。如果其他两个线程正在计算该结果，那么新到的线程将一直等待这个结果被计算出来。时间相比Memoizer2时间Memoizer3已经缩短了很多。  
  但是，仍然存在一个小缺陷，即在判断if(future == null)时，仍然可能有两个线程同时执行此处，判断f(1)时不在缓存中，将f(1)的Future放入缓存，再计算f(1)。因为判断是否将future是否存入缓存中的If代码块是非原子的“先检查再执行”，即当两个线程同时没有在缓存中找到期望的值，然后同一时间执行if()\{\}代码块。所以讲道理还没有严格实现判断其他线程是否同时开始计算。  
进一步改进，

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

运行结果：又快了180多，并发编程水很深哈。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 7]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70]: /images/20220325/11aed536f24f47399a7ca13c560ba53e.png
[20190121163827450.png]: /images/20220325/86686078f8484c9a9036be78d74ecf8a.png
[20190122110527680.png]: /images/20220325/e12b92240d4c49d49ea76fb2866b2150.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 1]: /images/20220325/7adc03c5b4e94537bf81ddf58bec46e0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 2]: /images/20220325/c715ad095bdc4ba3b340b4c67043c757.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 3]: /images/20220325/f6d70584128f4473b545b08fcd1ebb53.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 4]: /images/20220325/dc9c6e293c8846adbcf1eb1d15a43e34.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 5]: /images/20220325/7b385ee1fc644951a355c4df1bbc1d36.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 6]: /images/20220325/463e5759e31e4db0a89e14df75c0002c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 7]: /images/20220325/a3dae40c5cd14c11bbf3f0fa3b40af5e.png