《java并发编程实战》读书笔记4--基础构建模块，java中的同步容器类&并发容器类&同步工具类，消费者模式

Posted 2020-09-18

上一章说道委托是创建线程安全类的一个最有效策略，只需让现有的线程安全的类管理所有的状态即可。那么这章便说的是怎么利用java平台类库的并发基础构建模块呢？

---

5.1 同步容器类

包括Vector和Hashtable，此外还包括在JDK1.2中添加的一些功能相似的类，这些同步的封装器类由Collections.synchronizedXxx等工厂方法创建的。这些类实现线程安全的方式是：将他们的状态封装起来，并对每个共有方法都进行同步，使得每次只能有一个线程能访问容器的状态。

关于java中的Vector，看了下JDK文档，大致说的是实现了可变长数组功能和随机存取，继承的父类是AbstractList，也就是说Vector和ArrayList一样也是个List数据结构，区别在于ArrayList会比Vector快，但ArrayList是非同步的，如果设计涉及到多线程，还是用Vector比较好一些 。在JDK文档中也说明了这一点，Vector的迭代是是 fast-fail ，意思就是说如果返回了迭代器后，Vector的结构发生变化的话会抛出ConcurrentModificationException异常。基本的方法和List系列的集合差不多。关于Vector类用法的例子和Vector类的源码

关于Hashtable：java提高篇（二五）-----HashTable

-5.1.1 同步容器类的问题

就是说在使用同步容器类的时候某些情况下需要额外的客户端加锁来保护复合操作。容器上常见的复合操作包括：迭代以及运算条件，例如若没有则添加(检查Map中是否存在键值K，如果没有就加入二元组(K,V)).

同步容器类通过自身的锁来保护它的每个方法。通过获得容器类的锁，可以使getLast和deleteLast成为原子操作，如图：

在调用size和相应的get之间，Vector的长度可能会发生变化，这种风险在对Vector中的元素进行迭代时仍会出现，如图：

通过客户端加锁的方式解决这个问题，但是会导致其他线程在迭代期间无法访问它，因此降低了并发性。

-5.1.2 迭代器与ConcurrentModificationException

迭代器的“及时失败”的实现方式是：将计数器的变化和容器相关联起来，如果迭代期间计数器被修改，那么hasNext或next将抛出ConcurrentModificationException. 然而这种检查是在没有同步的情况下进行的，因此可能会看到失效的计数值，而迭代器可能并没有意识到已经发生了修改。单线程代码中也会抛出这个异常，当对象直接从容器中删除而不是通过Iterator.remove来删除时就会抛出这个异常。

下面是jdk中部分关于ArrayList的解释：

• Note that this implementation is not synchronized. If multiple threads access an ArrayList instance concurrently, and at least one of the threads modifies the list structurally, it must be synchronized externally. (A structural modification is any operation that adds or deletes one or more elements, or explicitly resizes the backing array; merely setting the value of an element is not a structural modification.) This is typically accomplished by synchronizing on some object that naturally encapsulates the list. If no such object exists, the list should be "wrapped" using the Collections.synchronizedList method. This is best done at creation time, to prevent accidental unsynchronized access to the list:

     List list = Collections.synchronizedList(new ArrayList(...));

为了并发性，如果不希望在迭代期间对容器加锁，一种替代方法是"克隆”容器，并在副本上进行迭代。

-5.1.3 隐藏迭代器

虽然加锁可以防止ConcurrentModificationException，但要记住在所有对共享容器进行迭代的地方都需要加锁，但是某些情况下迭代器会被隐藏起来，如：

 

5.2 并发容器类

java5.0增加了ConcurrentHashMap和CopyOnWriteArrayList，增加了两种新的容器类型：Queue和BlockingQueue。Queue用来临时保存一组等待处理的元素，它提供了几种实现，包括：ConcurrentLinkedQueue和PriorityQueue。

-5.2.1 ConcurrentHashMap

和HashMap一样，ConcurrentHashMap也是一个基于散列的Map，但是用一种完全不同的策略来提供更高的并发性和伸缩性。ConcurrentHashMap并不是将每个方法都在同一个锁上同步，不会使每次只能有一个线程访问容器，使用了分段锁机制，一种粒度更细的加锁机制。详细见ConcurrentHashMap总结

-5.2.2 额外的原子操作

由于ConcurrentHashMap不能被加锁来执行独占访问，因此无法使用客户端加锁来创建新的原子操作。可以考虑使用CurrentMap。关于CurrentMap

-5.2.3 CopyOnWriteArrayList

用于替代同步List。“写入时复制”容器的线程安全性在于，只要正确地发布了一个事实不可变对象，那么在访问该对象时就不需要进一步同步。在每次修改时，都会创建并重新发布一个新的容器副本，从而实现可变性。显然每当修改容器时都会复制底层数组。仅当迭代操作远远多于修改操作时，才应该使用"写入时复制"容器。这个准则很好的描述了事件通知系统：

关于CopyOnWriteArrayList

 

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

阻塞队列提供了可阻塞的put和take方法，以及支持定时的offer和poll方法。如果队列满了，那么put方法将阻塞直到有空间可用，如果队列为空，那么take方法将会阻塞直到有元素可用。BlockingQueue简化了生产者-消费者设计的实现过程。关于BlockingQueue。Java多线程（五）之BlockingQueue深入分析

-5.3.1 示例：桌面搜索

下面的程序生产者的任务是在某个文件层次结构中搜索符合索引标准的文件，并将它们的名称放入工作队列，消费者的任务是从队列中取出文件名称并对它们建立索引。

-5.3.2 串行线程封闭

-5.3.3 双端队列与工作密取

java6新增了两种容器类型, Deque和BlockingDeque，对应的是数据结构中的双端队列，实现了在队列头和队列尾的高效插入和移除。具体实现包括ArrayDeque和LinkedBlockingDeque。双端队列适用于另一种相关模式，即工作密取。在工作密取设计中，每个消费者都有各自的双端队列。如果一个消费者完成了自己双端队列中的全部工作，那么它可以从其他消费者双端队列末尾秘密地获取工作。

 

5.4 阻塞方法与中断方法

BlockingQueue的put和take等方法会抛出受检查异常(Checked Exception) Interrupted-Exception, 这与类库中其他一些方法的做法相同，例如Thread.sleep. 当某方法抛出InterruptedException时，表示该方法是一个阻塞方法。Thread提供了Interrupt方法，用于中断线程或者查询线程是否已经被中断。每个线程都有一个布尔类型的属性。当在代码中调用了一个将抛出InterruptedException方法时，你自己的方法也就变成了一个阻塞方法，必须要处理对中断的响应。有两种基本选择：传递InterruptedException和恢复中断

 

5.5 同步工具类

在容器类中，阻塞队列是一种独特的类，他们不仅能作为保存对象的容器，还能协调生产者和消费者等待线程的控制流。阻塞队列可以作为同步工具类，其它类型的同步工具类还包括信号量，栅栏以及闭锁

-5.5.1 闭锁

关于CountDownLatch

下面的程序给出了闭锁的两种常见的用法：

启动门将使得主线程能够同时释放所有的工作线程(而不是开一个线程就执行一个，这样先启动的线程将"领先"后启动的线程)，而结束门则使主线程能够等待最后一个线程执行完成。

-5.5.2 FutureTask

继续搬运博文：Java多线程编程：Callable、Future和FutureTask浅析（多线程编程之四）。

下面的程序使用了FutureTask来执行一个高开销的计算，并且计算结果将在稍后使用。通过提前启动计算，可以减少在等待结果时需要的时间。

-5.5.3 信号量

计数信号量(Counting semaphore)用来控制同时访问某个特定资源的操作数量，还可以用来实现某种资源池，或者对容器施加边界。Java多线程系列--“JUC锁”11之 Semaphore信号量的原理和示例

感觉这个原理是不是类似操作系统中的PV操作，那个好像也是什么信号量，不知道和java的这个是不是一回事。

-5.5.4 栅栏

前面提到的闭锁是一次性操作，一旦进入终止状态就不能被重置。Java并发学习笔记(四)-栅栏CyclicBarrier

另一种形式的栅栏是Exchanger，它是一种两方栅栏，各方在栅栏位置上交换数据。Java Thread&Concurrency(4): 深入理解Exchanger实现原理

 

5.6 例子: 构建高效且可伸缩的结果缓存

本节将开发一个高效且可伸缩的缓存，用于改进一个高计算开销的函数。首先从HashMap开始，然后分析它的并发性缺陷，并讨论如何修复它们。

接着我们用ConcurrentHashMap来替换HashMap：

 

下面的程序中的Memoizer3将用于缓存值的Map重新定义为ConcurrentHashMap<A, Future<V>>,替换原来的ConcurrentHashMap<A,V>. 关于java中的Future：Java线程(七)：Callable和Future. Future是一个接口，FutureTask是它的实现。

现在的设计仍然存在一个缺陷：仍然存在两个线程计算出相同值的漏洞(就是重复计算啦)。由于compute方法中的if代码块仍然是非原子的"先检查再执行"操作，因此两个线程仍有可能在同一时间内调用compute来计算相同的值。如下图所示：

这个问题的原因是：复合操作("若没有则添加")是在底层的Map对象上执行的(就是那个cache.put(arg,ft))，而这个对象无法通过加锁来确保原子性。下面的程序解决了这个问题：

当缓存的是Future而不是值时，将导致缓存污染问题：如果某个计算被取消或者失败，那么在计算这个结果时将指明计算过程被取消或者失败。为了解决这种情况，如果Memoizer发现计算被取消或检测带RuntimeException，就移除Future。Memoizer同样也没有解决缓存逾期的问题，但可以通过使用FutureTask的子类来解决，在子类中为每个结果指定一个逾期时间，并定期扫描缓存中的逾期元素。同样，缓存清理问题也要去解决。

 

---

小结：