Concurrent In Java 6 分享，第一部分 集合

第一部分 集合 http://jimichan.iteye.com/blog/951948

 

第二部分 线程池 http://jimichan.iteye.com/blog/951950

 

第三部分 锁 http://jimichan.iteye.com/blog/951954

 

第四部分 同步辅助类 http://jimichan.iteye.com/blog/951955

Concurrent In Java，第一部分 集合

 

2011-3-9  延昭 & 陈汝烨 版权所有，特别禁止发布到百度文库

 

这篇是来自公司内部分享会议是写的总结，有些内容没有表达出来，大家可以来踩，但是需留下原因，以便后续补充。

这一系列只是对JUC各个部分做了说明和介绍，没人深入原理！

concurrent并发包，让你易于编写并发程序。并发下我们经常需要使用的基础设施和解决的问题有ThreadPool、Lock、管道、集合点、线程之间等待和唤醒、线程间数据传输、共享资源访问控制、并发线程之间的相互等待，等待。

 

concurrent提供的工具能够解决绝大部分的场景，还能提高程序吞吐量。

 

现代的服务器多采用多核CPU，从而不同线程之间有可能真正地在同时运行而不是cpu时间切片。在处理大计算量的程序上要尽可能利用CPU多核特性，提高系统吞吐量。

 

并发编程主要面临三个问题：

1.如何让多个线程同时为同一个任务工作（并发编程设计）

2.多个线程之间对共享资源的争用。

3.多个线程之间如何相互合作、传递数据。

 

1. concurrent包提供的集合

concurrent包直接提供了标准集合的一些实现，在下面做简单介绍。在大部分情况下可以使用它们提供高并发环境下对集合访问的吞吐量。

 

1.1 ConcurrentHashMap

Map的一个并发实现。在多线程环境下，它具有很高的吞吐量和具备可靠的数据一致性。它支持并发读和一定程度的并发修改（默认16个并发，可以通过构造函数修改）。

 

HashMap的实现是非线程安全的，高并发下会get方法常会死锁，有的时候会表现为CPU居高不下。

   public V get(Object key) {

       if (key == null)

           return getForNullKey();

       int hash = hash(key.hashCode());

       for (Entry<K,V> e = table[indexFor(hash, table.length)];

            e != null;

            e = e.next) {

           Object k;

           if (e.hash == hash && ((k = e.key) == key || key.equals(k)))

               return e.value;

       }

       return null;

   }

在get操作里面for循环取对象的操作，由于高并发同时读写，for循环的结果变得不可预知，所以有可能一直循环。

所以高并发环境下尽量不要直接使用HashMap，对系统造成的影响很难排除。

 

 

和Collections.synchronizedMap(new HashMap(...))相比，外ConcurrentHashMap在高并发的环境下有着更优秀的吞吐量。因为ConcurrentHashMap可以支持写并发，基本原理是内部分段，分段的数量决定着并发程度。通过concurrencyLevel参数可以设置。如果你能预期并发数量那么设置该参数可以获取更优吞吐量。

 

另外为ConcurrentHashMap还实现了：

 

V putIfAbsent(K key, V value);

boolean remove(Object key, Object value);

boolean replace(K key, V oldValue, V newValue);

V replace(K key, V value);

 

这四个一致性的操作方法。

 

1.2 BlockingQueue

BlockingQueue定义了一个接口，继承了Queue接口。Queue是一种数据结构，意思是它的项以先入先出（FIFO）顺序存储。

 

BlockingQueue为我们提供了一些多线程阻塞语义的方法，新增和重定义了一些方法插入：

 

	抛出异常	返回的布尔值	阻塞	超时
插入	add(e)	offer(e)	put(e)	offer(e, time, unit)
移除	remove()	poll()	take()	poll(time, unit)
检查	element()	peek()

 

BlockingQueue是线程安全的，非常适合多个生产者和多个消费者线程之间传递数据。

 

形象地理解，BlockingQueue好比有很多格子的传输带系统，不过当你（生产者）调用put方法的时候，如果有空闲的格子那么放入物体后立刻返回，如果没有空闲格子那么一直处于等待状态。add方法意味着如果没有空闲格子系统就会报警，然后如果处理该报警则按照你的意愿。offer方法优先于add方法，它通过返回true 或 flase来告诉你是否放入成功。offer超时方法，如果不空闲的情况下，尝试等待一段时间。

 

BlockingQueue有很多实现ArrayBlockingQueue, DelayQueue, LinkedBlockingDeque, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue

 

补充Dueue是个双向队列，可以当做堆栈来使用。

 

BlockingQueue在ThreadPool中，作为任务队列来使用，用来保存没有立刻执行的工作任务对象。

1.3 SynchronousQueue

SychronousQueue是BlockingQueue的一个实现，它看起来是一个队列，但是其实没有容量，是特定条件下的一个精简实现。

 

做个比喻，SychronousQueue对象就像一个接力棒，现在有两个运动员交棒者和接棒者（线程）要做交接。在交接点，交棒者没有交出之前是不能松开的（一种等待状态），接棒者在接到棒之前是必须等待。换一句话说不管谁先到交接点，必须处于等待状态。

 

在生产者和消费者模型中。如果生产者向SychronousQueue进行put操作，直到有另外的消费者线程进行take操作时才能返回。对消费者也是一样，take操作会被阻塞，直到生产者put。

 

在这种生产者-消费者模型下，生产者和消费者是进行手对手传递产品，在消费者消费一个产品之前，生产者必须处于等待状态。它给我们提供了在线程之间交换单一元素的极轻量级方法，并且具有阻塞语义。

 

提示：上面举例中有写局限性。其实生产者和消费者进程是可以任意数量的。M：N。生产线程之间会对SychronousQueue进行争用，消费者也是一样。

 

对SychronousQueue类似于其他语境中“会合通道”或 “连接”点问题。它非常适合于传递性设计，在这种设计中，在一个线程中运行的对象要将某些信息、事件或任务传递给在另一个线程中运行的对象，它就必须与该对象同步。

 

1.4Exchanger

是SychronousQueue的双向实现。用来伙伴线程间交互对象。Exchanger 可能在比如遗传算法和管道设计中很有用。

 

形象地说，就是两个人在预定的地方交互物品，任何一方没到之前都处于等待状态。

1.5 CopyOnWriteArrayList 和 CopyOnWriteArraySet

它们分别是List接口和Set接口的实现。正如类名所描述的那样，当数据结构发生变化的时候，会复制自身的内容，来保证一致性。大家都知道复制全部副本是非常昂贵的操作，看来这是一个非常不好的实现。事实上没有最好和最差的方案，只有最合适的方案。一般情况下，处理多线程同步问题，我们倾向使用同步的 ArrayList，但同步也有其成本。

 

那么在什么情况下使用CopyOnWriteArrayList 或者CopyOnWriteArraySet呢？

1. 数据量小。
2. 对数据结构的修改是偶然发生的，相对于读操作。

 

举例来说，如果我们实现观察者模式的话，作为监听器集合是非常合适的。

 

1.6 TimeUnit

 

虽然是个时间单位，但是它也是concurrent包里面的。也许你以前的代码里面经常出现1*60*1000来表示一分钟，代码可读性很差。现在你可以通过TimeUnit来编写可读性更好的代码，concurrent的api里面涉及到时间的地方都会使用该对象。

 

 

我之所以先进并发框架常用的集合，是因为线程池的实现特性都利用了BlockingQueue的一些特性。

 

 

 

请继续收看 第二部分 ThreadPool

 http://jimichan.iteye.com/blog/951950