第6章 Java并发容器和框架

6.1  ConcurrentHashMap（线程安全的HashMap、锁分段技术）

6.1.1 为什么要使用ConcurrentHashMap

　　在并发编程中使用HashMap可能导致程序死循环，而线程安全的HashTable效率又非常低下。基于以上两个原因，便有了ConcurrentHashMap的登场机会。

　　（1）线程不安全的HashMap

　　在多线程环境下，使用HashMap进行put操作会引起死循环（因为多线程会导致HashMap的Entry链表形成环形数据结构，Entry的next节点永远不为空，就会产生死循环获取Entry。），导致CPU利用率接近100%，所以在并发情况下不能使用HashMap。

　　（2）效率低下的HashTable

　　HashTable容器使用sychronized来保证线程安全，但在线程竞争激烈的情况下HashTable的效率非常低下。因为当一个线程访问HashTable的同步方法，其他线程也访问HashTable的同步方法时，会进入阻塞或轮询状态。如线程1使用put进行元素添加，线程2不但不能用put方法添加元素，也不能使用get方法来获取元素，所以竞争越激烈效率越低。

　　（3）ConcurrentHashMap的锁分段技术可有效提升并发访问率

　　HashTable容器在竞争激烈的并发环境下表现效率低下的原因是所有访问HashTable的线程都必须竞争同一把锁。假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效提高并发访问效率，这就是ConcurrentHashMap所使用的锁分段技术。首先将数据分成一段一段地存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一段数据的时候，其他段的数据也能被其他线程访问。

ConcurrentHashMap的结构：

　　ConcurrentHashMap由Segment数组结构和HashEntry数组结构组成。Segment是可重入锁，扮演锁的角色；HashEntry存储键值对数据。
　　一个ConcurrentHashMap里包含一个Segment数组，Segment的结构与HashMap类似，是一种数组和链表结构。一个Segment包含一个HashEntry数组，每个HashEntry是一个链表结构的元素，每个Segment守护一个HashEntry数组里的元素。当对HashEntry数组的数据进行修改的时候，必须首先获得与它对应的Segment锁。

ConcurrentHashMap的操作：

　　get操作：get过程不需要加锁，只有值为空值的时候才加锁重读。（如何做到不加锁的？get方法里将要使用的共享变量都定义为volatile类型。）

　　put操作：put过程必须加锁（由于put方法里需要对共享变量进行写入操作，所以为了线程安全，在操作共享变量时必须加锁）。put方法首先定位到Segment，然后在segment里进行插入操作。

　　插入操作步骤：第一步判断是否需要对Segment里的HashEntry数组进行扩容，第二步定位添加元素的位置，然后将其放到HashEntry数组里。

• 是否需要扩容？ 在插入元素前先判断Segment里的HashEntry数组是否超过容量（threadshold），如果超过阈值，则对数组进行扩容。
• 如何扩容？ 在扩容时，首先会创建一个容量为原来容量2倍的数组，然后将原数组里的元素进行再散列后插入到新的数组里。为了高效，ConcurrentHashMap不会对整个容器进行扩容，而只对某个segment进行扩容。

size操作：先尝试2次通过不锁住Segment的方式来统计各个Segment大小，如果统计过程中count发生了变化，则再采用加锁的方式（统计size的时候把所有Segment的put、remove、clean方法全部锁住）来统计所有Segment的大小。

• ConcurrentHashMap如何判断在统计的时候容器是否发生了变化呢？ 使用modCount变量，在put、remove和clean方法里操作元素前都会将变量modCount进行加1，那么在统计size前后比较modCount是否发生变化，从而得知容器的大小是否发生变化。

6.2  ConcurrentLinkedQueue（非阻塞的线程安全队列）

　　实现一个线程安全的队列有两种方式：

• 使用阻塞方法：用一个锁（入队和出队用同一把锁）或者用两个锁（入队和出队用不同的锁）等方式实现。
• 使用非阻塞的方法：使用循环CAS。

　　ConcurrentLinkedQueue是一个基于链接结点的*线程安全队列，采用“先进先出”规则对节点进行排序。它采用了“wait-free”算法（即CAS算法）来实现。

6.3  阻塞队列

　　阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作支持阻塞的插入和移除方法。

• 支持阻塞的插入方法：当队列满时，队列会阻塞插入元素的线程，直到队列不满。
• 支持阻塞的移除方法：当队列空时，获取元素的线程会等待队列变为非空。

　　阻塞队列常用于生产者和消费者的场景：生产者是向队列里添加元素的线程，消费者是从队列里取元素的线程。阻塞队列就是生产者用来存放元素、消费者用来获取元素的容器。

　　在阻塞队列不可应时，这两个附加操作（插入和移除）的4种处理方式：

方法/处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put(e)	offer(e, time, unit)
移除方法	remove()	poll()	take()	poll(time, unit)
检查方法	element()	peek()	不可用	不可用

• 抛出异常    ：当队列满时，如果再往队列里插入元素会抛出IllegalStateException("Queue full")异常。当队列空时，从队列里获取元素会抛出NoSuchElementException异常。
• 返回特殊值：当往队列插入元素时，会返回元素是否插入成功，成功返回true。如果是移除方法，则从队列里取出一个元素，如果没有则返回null。
• 一直阻塞   ：当阻塞队列满时，如果生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到队列可用或者响应中断退出。当队列空时，如果消费者线程从队列列take元素，队列会阻塞消费者线程，直到队列不为空。
• 超时退出   ：当阻塞队列满时，如果生产者线程往队列里插入元素，队列会阻塞生产者线程一段时间，如果超时则退出。

【注】：如果是*阻塞队列，队列不可能会出现满的情况，所以使用put或offer方法永远不会被阻塞，而且使用offer方法时，该方法永远返回true。

JDK7提供了7个阻塞队列：

• ArrayBlockingQueue   ：数组结构组成的有界阻塞队列，按FIFO原则对元素进行排序。
• LinkedBlockingQueue ：链表结构组成的有界阻塞队列，默认和最大长度为Integer.MAX_VALUE，按FIFO原则对元素进行排序。
• PriorityBlockingQueue：支持优先级的*阻塞队列，默认情况下元素采取自然顺序升序排列。不保证同优先级元素的顺序。
• DelayQueue                  ：支持延时获取元素的*阻塞队列。队列使用PriorityQueue实现，队列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。可应用于：
  • 缓存系统的设计：用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素表示缓存有效期到了。
  • 定时任务调度：使用DelayQueue保存当天将会执行的任务和执行时间，一旦从DelayQueue中获取到任务就开始执行，比如TimerQueue就是使用DelayQueue实现的。
• SynchronousQueue     ：不存储元素的阻塞队列。每一个put操作必须等待一个take操作，否则不能继续添加元素。
• LinkedTransferQueue  ：链表结构组成的*阻塞TransferQueue队列。相对于其他阻塞队列，LinkedTransferQueue多了tryTransfer和transfer方法。
  • transfer方法：如果当前有消费者正在等待接收元素，transfer方法可以把生产者传入的元素立刻传给消费者。如果没有消费者在等待接收元素，则将元素存放在队列tail节点并等到钙元素被消费者消费了才返回。
  • tryTransfer方法：如果没有消费者等待接收元素，则立即返回false。
• LinkedBlockingDeque ：链表结构组成的双向阻塞队列。可以从队列的两端插入和移出元素。

阻塞队列的实现原理：

　　如果队列是空的，消费者会一直等待，当生产者添加元素时，消费者是如何知道当前队列有元素的呢？

　　使用通知模式实现。就是当生产者往满的队列里添加元素时会阻塞住生产者，当消费者消费了一个队列中的元素后，会通知生产者当前队列可用。

6.4  Fork/Join框架

• Fork/Join框架是一个用于并行执行任务的框架，是一个把大任务分隔成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。

• 工作窃取算法：是指某个线程从其他队列里窃取任务拉执行。假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。
  • 工作窃取算法的优点是充分利用线程进行并行计算，并减少了线程间的竞争，其缺点是在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且消耗了更多的系统资源，比如创建多个线程和多个双端队列。

Fork/Join框架的设计：

　　步骤1：分割任务。首先需要有一个fork类来把大任务分割成子任务，有可能子任务还是很大，所以还需要不停地分割，直到分割出的子任务足够小。

　　步骤2：执行任务并合并结果。分割的子任务分别放在双端队列里，然后几个启动线程分别从双端队列里获取任务执行。子任务执行完的结果都统一放在一个队列里，启动一个线程从队列里拿数据，然后合并这些数据。

　　Fork/Join使用两个类来完成以上两件事情：

• ForkJoinTask：我们要使用ForkJoin框架，必须首先创建一个ForkJoin任务。它提供在任务中执行fork()和join()操作的机制，通常情况下我们不需要直接继承ForkJoinTask类，而只需要继承它的子类，Fork/Join框架提供了以下两个子类：
  • RecursiveAction：用于没有返回结果的任务。
  • RecursiveTask ：用于有返回结果的任务。
• ForkJoinPool ：ForkJoinTask需要通过ForkJoinPool来执行。
  • ForkJoinPool由ForkJoinTask数组和ForkJoinWorkerThread数组组成，ForkJoinTask数组负责将存放程序提交给ForkJoinPool的任务，而ForkJoinWorkerThread数组负责执行这些任务。

　　任务分割出的子任务会添加到当前工作线程所维护的双端队列中，进入队列的头部。当一个工作线程的队列里暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务。

Fork/Join框架的异常处理：

　　ForkJoinTask在执行的时候可能会抛出异常，但是我们没办法在主线程里直接捕获异常，所以ForkJoinTask提供了isCompletedAbnormally()方法来检查任务是否已经抛出异常或已经被取消了，并且可以通过ForkJoinTask的getException方法获取异常。其中，getException方法返回Throwable对象，如果任务被取消了则返回CancellationException，如果任务没有完成或者没有抛出异常则返回null。

第7章  Java中的13个原子操作类

　　当程序更新一个变量时，如果多线程同时更新这个变量，可能得到期望值之外的值。通常我们使用sychronized来解决这个问题，sychronized会保证多线程不会同时更新同一个变量。

　　而Java从JDK1.5开始提供了java.util.concurrent.atomic包，包中的原子操作类提供了一种用法简单、性能高效、线程安全地更新一个变量的方式。一共提供了13个类，属于4种类型的原子更新方式，分别是原子更新基本类型、原子更新数组、原子更新引用和原子更新属性（字段）。

（1）原子更新基本类型

• AtomicBoolean ：原子更新布尔类型
• AtomicInteger： 原子更新整型
• AtomicLong: 原子更新长整型

（2）原子更新数组

• AtomicIntegerArray ：原子更新整型数组里的元素
• AtomicLongArray :原子更新长整型数组里的元素
• AtomicReferenceArray : 原子更新引用类型数组的元素

（3）原子更新引用类型

• AtomicReference ：原子更新引用类型
• AtomicReferenceFieldUpdater ：原子更新引用类型里的字段
• AtomicMarkableReference：原子更新带有标记位的引用类型。可以原子更新一个布尔类型的标记位和应用类型

（4）原子更新字段类

• AtomicIntegerFieldUpdater:原子更新整型的字段的更新器
• AtomicLongFieldUpdater：原子更新长整型字段的更新器
• AtomicStampedReference:原子更新带有版本号的引用类型。该类将整型数值与引用关联起来，可用于原子的更新数据和数据的版本号，可以解决使用CAS进行原子更新时可能出现的ABA问题。

【注】要想原子地更新字段类需要2步。第一步：因为原子更新字段类都是抽象类，每次使用的时候必须使用静态方法newUpdate()创建一个更新器，并且需要设置想要更新的类和属性。第二步：更新类的字段（属性）必须使用public volatile修饰符。

第8章 Java中的并发工具类

（1）等待多线程完成的CountDownLatch

　　CountDownLatch允许一个或多个线程等待其他线程完成操作。

　　要实现主线程等待所有线程完成某个操作，最简单的做法是使用join()方法。join()用于让当前执行线程等待join线程执行结束。其实现原理是不停检查join线程是否存活，如果join线程存活则让当前线程永远等待。直到join线程中止后，线程的this.notifyAll()方法会被调用。

　　在JDK1.5之后的并发包中提供的CountDownLatch也可以实现join的功能。

• CountDownLatch的构造函数接收一个int类型的参数作为计数器，如果你想等待N个点（N个线程）完成，就传入N。  static CountDownLatch c = new CountDownLatch(2);
• 每次调用CountDownLatch的countDown方法时，N就减1，CountDownLatch的await方法会阻塞当前线程，直到N变成0。

（2）同步屏障CyclicBarrier

　　CyclicBarrier的作用是：让一组线程到达一个屏障（也可以称之为同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才能继续运行。

• CyclicBarrier(int parties)构造函数接收一个int参数用来设置拦截线程的数量，还有一个更高级的构造函数CyclicBarrier(int parties, Runnable barrierAction)，用于在线程到达屏障时，优先执行barrierAction。  static CyclicBarrier c = new CyclicBarrier(2，new A);

CountDownLatch与CyclicBarrier的区别：

　　CountDownLatch的计数器只能使用1次，而CyclicBarrier的计数器可以使用reset()方法重置。（所以CyclicBarrier能处理更为复杂的业务场景。例如，如果计算发生错误，可以重置计数器，并让线程重新执行1次。）

（3）控制并发线程数的Semaphore

　　Semaphore(信号量)用来控制同时访问特定资源的线程数量，通过协调各个线程以保证合理的使用公共资源。

　　应用场景：可以用于做流量控制，特别是共有资源有限的应用场景，比如数据库连接。

• Semaphore(int permits)构造方法接收一个int参数，表示可用的许可证数量。
• 每次线程使用Semaphore的acquire()方法获取一个许可证，用完后调用release()方法归还

（4）线程间交换数据的Exchanger

　　Exchanger（交换者）是一个用于线程间协作的工具类。用于进行线程间的数据交换。 它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据，如果第一个线程先执行exchange()方法，它会一直等待第二个线程也执行exchange()方法，当两个线程到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。

　　如果两个线程有一个没有执行exchange()方法，则会一直等待，如果担心有特殊情况发生，避免一直等待，可以使用exchange(V x, longtimeout , TimeUnit unit)设置最大等待时长。