【转】JAVA并发容器代码随读

转载自 http://rdc.taobao.com/team/jm/archives/539

 

队列类型的BlockingQueue和 ConcurrentLinkedQueue（生产者-消费者模型）,
Map类型的ConcurrentMap,
Set类型的ConcurrentSkipListSet和 CopyOnWriteArraySet,
List类型的CopyOnWriteArrayList.
并发环境性能优于Vector、Hashtable

1. java.util.concurrent所提供的并发容器
         java.util.concurrent提供了多种并发容器，总体上来说有4类，队列类型的BlockingQueue和 ConcurrentLinkedQueue,Map类型的ConcurrentMap,Set类型的ConcurrentSkipListSet和CopyOnWriteArraySet,List类型的CopyOnWriteArrayList.

           这些并发容器都采用了多种手段控制并发的存取操作，并且尽可能减小控制并发所带来的性能损耗。接下来我们会对每一种类型的实现类进行代码分析，进而得到java.util.con current包所提供的并发容器在传统容器上所做的工作。

2. BlockingQueue
           BlockingQueue接口定义的所有方法实现都是线程安全的，它的实现类里面都会用锁和其他控制并发的手段保证这种线程安全，但是这些类同时也实现了Collection接口(主要是AbstractQueue实现)，所以会出现BlockingQueue的实现类也能同时使用Conllection接口方法，而这时会出现的问题就是像addAll,containsAll,retainAll和removeAll这类批量方法的实现不保证线程安全，举个例子就是addAll 10个items到一个ArrayBlockingQueue,可能中途失败但是却有几个item已经被放进这个队列里面了。

下面我们根据这幅类图来逐个解析不同实现类的特性和特性实现代码

         DelayQueue提供了一个只返回超时元素的阻塞队列，也就是说，即使队列中已经有数据了，但是poll或者take的时候还要判定这个element有没达到规定的超时时间,poll方法在element还没达到规定的超时时间返回null,take则会通过condition.waitNanos()进入等待状态。一般存储的element类型为Delayed，这个接口JDK中实现的类有ScheduledFutureTask,而DelayQueue为DelayedWorkQueue的Task容器，后者是ScheduledThreadPoolExecutor的工作队列，所以DelayQueue所具有的超时提供元素和线程安全特性对于并发的定时任务有很大的意义。

public E take() throws InterruptedException {

        final ReentrantLock lock = this.lock;

        //控制并发

        lock.lockInterruptibly();

        try {

            for (;;) {

                E first = q.peek();

                if (first == null) {

                    //condition协调队列里面元素

                    available.await();

                } else {

                    long delay =  first.getDelay(TimeUnit.NANOSECONDS);

                    if (delay > 0) {

                         //因为first在队列里面的delay最短的(优先队列保证),所以wait这个时间那么队列中最短delay的元素就超时了.即

                        //队列有元素供应了.

                        long tl = available.awaitNanos(delay);

                    } else {

                        E x = q.poll();

                        assert x != null;

                        if (q.size() != 0)

                            available.signalAll(); // wake up other takers

                        return x;

                    }

                }

            }

        } finally {

            lock.unlock();

        }

}

        DelayQueue的内部数据结构是PriorityQueue,因为Delayed接口同时继承了Comparable接口，并且Delayed的实现类对于这个compareTo方法的实现是基于超时时间进行大小比较，所以DelayQueue无需关心数据的排序问题，只需要做好存取的并发控制（ReetranLock）和超时判定即可。另外，DelayQueue有一个实现细节就是通过一个Condition来协调队列中是否有数据可以提供，这对于take和带有提取超时时间的poll是有意义的（生产者，消费者的实现）。

         PriorityBlockingQueue实现对于外部而言是按照元素的某种顺序返回元素，同时对存取提供并发保护(ReetranLock),使用Condition协调队列是否有新元素提供。PriorityBlocking Queue内部的数据结构为PriorityQueue,优先级排序工作交给PriorityQueue，至于怎么排序，需要根据插入元素的Comparable的接口实现，和DelayQueue比起来，它没有限定死插入数据的Comparable实现，而DelayQueue的元素实现Comparable必须按照超时时间的长短进行比较，否则DelayQueue返回的元素就很可能是错误的。

            ArrayBlockingQueue是一个先入先出的队列，内部数据结构为一个数组，并且一旦创建这个队列的长度是不可改变的，当然put数据时，这个队列也不会自动增长。ArrayBlockingQueue也是使用ReetranLock来保证存取的原子性，不过使用了notEmpty和notFull两个Condition来协调队列为空和队列为满的状态转换，插入数据的时候，判定当前内部数据结构数组E[] items的长度是否等于元素计数，如果相等，说明队列满，notFull.await()，直到items数组重新不为满（removeAt,poll等），插入数据后notEmpty.sinal()通知所有取数据或者移除数据并且因为items为空而等待的线程可以继续进行操作了。提取数据或者移除数据的过程刚好相反。

            ArrayBlockingQueue使用三个数字来维护队列里面的数据变更,包括takeIndex,putIndex,count,这里需要讲一下 takeIndex和putIndex,其中takeIndex指向下一个能够被提取的元素,而putIndex指向下一个能够插入数据的位置,实现类似下图的结构,当takeIndex移到内部数组items最大长度时,重新赋值为0,也就是回到数组头部,putIndex也是相同的策略.

/**

* 循环增加putIndex和takeIndex,如果到数组尾部,那么

* 置为0

*/

final int inc(int i) {

    return (++i == items.length)? 0 : i;

}

//**

* 插入一个item,需要执行线程获得了锁

*/

private void insert(E x) {

    items[putIndex] = x;

    //累加putIndex,可能到数组尾部,那么重新指向0位置

    putIndex = inc(putIndex);

    ++count;

     //put后,使用Condition通知正在等待take的线程可以做提取操作

    notEmpty.signal();

}

/**

* 获取一个元素,执行这个操作的前提是线程已经获得锁,

* 内部调用

*/

private E extract() {

    final E[] items = this.items;

    E x = items[takeIndex];

    items[takeIndex] = null;

    //累加takeIndex,有可能到数组尾部,重新调到数组头部

    takeIndex = inc(takeIndex);

    --count;

    //take后,使用Condition通知正在等待插入的线程可以插入

    notFull.signal();

    return x;

}

这里需要解释下Condition的实现，Condition现在的JDK实现只有AQS的ConditionObject，并且通过ReetranLock的newConditon()方法暴露出来，这是因为Condition的await()或者sinal()一般在lock.lock()与lock.unlock()之间执行，当执行condition.await()方法时，它会首先释放掉本线程持有的锁，然后自己进入等待队列，直到sinal(),唤醒后又会重新去试图拿到锁,拿到后执行await下方的代码,其中释放当前锁和得到当前锁都需要ReetranLock的tryAcquire（int args）方法来判定，并且享受ReetranLock的重进入特性。

public final void await() throws InterruptedException {

    if (Thread.interrupted())

        throw new InterruptedException();

    //加一个新的condition等待节点

    Node node = addConditionWaiter();

    //释放自己占用的锁

    int savedState = fullyRelease(node);

    int interruptMode = 0;

    while (!isOnSyncQueue(node)) {

        //如果当前线程等待状态是CONDITION,park住当前线程,等待condition的signal来解除

        LockSupport.park(this);

        if ((interruptMode =checkInterruptWhileWaiting(node)) != 0)

            break;

    }

    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)

        interruptMode = REINTERRUPT;

    if (node.nextWaiter != null)

        unlinkCancelledWaiters();

    if (interruptMode != 0)

        reportInterruptAfterWait(interruptMode);

}

            LinkedBlockingQueue是一个链表结构构成的队列，并且节点是单向的，也就是只有next,没有prev,可以设置容量，如果不设置，最大容量为Integer.MAX_VALUE，队列只持有头结点和尾节点以及元素数量，通过putLock和takeLock两个ReetranLock分别控制存和取的并发,但是remove,toArray,toString,clear, drainTo以及迭代器等操作会同时取得putLock和takeLock，并且同时lock,此时存或者取操作都会不可进行，这里有个细节需要注意的就是所有需要同时lock的地方顺序都是先putLock.lock再takeLock.lock，这样就避免了可能出现的死锁问题。takeLock实例化出一个notEmpty的Condition,putLock实例化一个notFull的Condition,两个Condition协调即时通知线程队列满与不满的状态信息，这在前面几种BlockingQueue实现中也非常常见，在需要用到线程间通知的场景时，各位不妨参考下。另外dequeue的时候需要改变头节点的引用地址,否则肯定会造成不能GC而内存泄露

private E dequeue() {

    Node<E> h = head;

    Node<E> first = h.next;

    //将原始节点的next指针指向自己,这样就能GC到自己否则虚拟机会认为这个节点仍然在用而不销毁(不知道是否理解有误)

    h.next = h; // help GC

    head = first;

    E x = first.item;

    first.item = null;

    return x;

}

           BlockingDequeue为阻塞的双端队列接口，继承了BlockingQueue，双端队列的最大的特性就是能够将元素添加到队列末尾,也能够添加到队列首部，取元素也是如此。LinkedBlockingDequeue实现了BlockingDequeue接口，就像LinkedBlockingQueue类似，也是由链表结构构成，但是和LinkedBlockingQueue不一样的是，节点元素变成了可双向检索，也就是一个Node持有next节点引用，同时持有prev节点引用，这对队列的头尾数据存取是有决定性意义的。LinkedBlockingDequeue只采用了一个ReetranLock来控制存取并发，并且由这个lock实例化了2个Condition notEmpty和notFull,count变量维护队列长度,这里只使用一个lock来维护队列的读写并发,个人理解是头尾的读写如果使用头尾分开的2个锁,在维护队列长度和队列Empty/Full状态会带来问题,如果使用队列长度做为判定依据将不得不对这个变量进行锁定.

//无论是offerLast,offerFirst,pollFirst,pollLast等等方法都会使用同一把锁.

public E pollFirst() {

    final ReentrantLock lock = this.lock;

    lock.lock();

    try {

        return unlinkFirst();

    } finally {

        lock.unlock();

    }

}

public E pollLast() {

    final ReentrantLock lock = this.lock;

    lock.lock();

    try {

        return unlinkLast();

    } finally {

        lock.unlock();

    }

 }

3. ConcurrentMap
             ConcurrentMap定义了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)四个方法，几个方法的特性并不难理解，4个方法都是线程安全的。

            ConcurrentHashMap是ConcurrentMap的一个实现类，这个类的实现相当经典，基本思想就是分拆锁，默认ConcurrentHashMap会实例化一个持有16个Segment对象的数组，Segment数组大小是可以设定的,构造函数里的concurrencyLevel指定这个值,但是需要注意的是,这个值并不是直接赋值.Segment数组最大长度为MAX_SEGMENTS = 1 << 16

int sshift = 0;

int ssize = 1;

//ssize是左移位的,也就是2,4,8,16,32增长(*2),所以你设定concurrencyLevel为10的时候,这个时候并发数最大为8.

while (ssize < concurrencyLevel) {

    ++sshift;

    ssize <<= 1;

}

每个Segment维持一个自动增长的HashEntry数组(根据一个阈值确定是否要增长长度，并不是满了才做).

int c = count;

//threshold一般(int)(capacity * loadFactor),

if (c++ > threshold)

    rehash();

下面3段代码是ConcurrentHashMap的初始化Segment,计算hash值,以及如何选择Segment的代码以及示例注解.

public ConcurrentHashMap(int initialCapacity,

        float loadFactor, int concurrencyLevel) {

    if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)

        throw new IllegalArgumentException();

    //首先确定segment的个数,左移位,并且记录移了几次,比如conurrencyLevel为30,那么2->4->8->16,ssize为16,sshift为4

    if (concurrencyLevel > MAX_SEGMENTS)

        concurrencyLevel = MAX_SEGMENTS;

    int sshift = 0;

    int ssize = 1;

    while (ssize < concurrencyLevel) {

        ++sshift;

        ssize <<= 1;

    }

    //segmentShift为28

    segmentShift = 32 - sshift;

    //segmentMask为15

    segmentMask = ssize - 1;

    //this.segments=new Segment[16]

    this.segments = Segment.newArray(ssize);

    if (initialCapacity > MAXIMUM_CAPACITY)

        initialCapacity = MAXIMUM_CAPACITY;

    //假设initialCapacity使用32,那么c=2

    int c = initialCapacity / ssize;

    if (c * ssize < initialCapacity)

        ++c;

    int cap = 1;

    //cap为2

    while (cap < c)

        cap <<= 1;

    //每个Segment的容量为2

    for (int i = 0; i < this.segments.length; ++i)

        this.segments[i] = new Segment<K,V>(cap, loadFactor);

}

/** *//**

 *segmentShift为28,segmentMask为15(1111)

 *因为hash值为int,所以32位的

 *hash >>> segentShift会留下最高的4位,

 *再与mask 1111做&操作

 *所以这个最终会产生 0-15的序列.

 */

final Segment<K,V> segmentFor(int hash) {

    return segments[(hash >>> segmentShift) & segmentMask];

}

/**

 *将计算的hash值补充到原始hashCode中,这是为了防止

   *外部用户传进来劣质的hash值(比如重复度很高)所带来

   *的危害.

 */

private static int hash(int h) {

    // Spread bits to regularize both segment and index locations,

    // using variant of single-word Wang/Jenkins hash.

    h += (h <<  15) ^ 0xffffcd7d;

    h ^= (h >>> 10);

    h += (h <<   3);

    h ^= (h >>>  6);

    h += (h <<   2) + (h << 14);

    return h ^ (h >>> 16);

}

            当put进来一个key、value对，ConcurrentHashMap会计算Key的hash值，然后从Segment数组根据key的Hash值选出一个Segment，调用其put方法，Segment级别的put方法通过ReetranLock来控制读取的并发，其实Segment本身继承了ReetranLock类。

                Segment的put方法在lock()后，首先对数组长度加了新的元素之后是否会超过阈值threshold进行了判定，如果超过，那么进行rehash()，rehash()的过程相对繁琐，首先数组会自动增长一倍，然后需要对HashEntry数组中的所有元素都需要重新计算hash值，并且置到新数组的新的位置，同时为了减小操作损耗，将原来不需要移动的数据不做移动操作(power-of-two expansion,在默认threshold,在数组扩大一倍时只需要移动1/6元素,其他都可以不动)。所有动作完成之后，通过一个while循环寻找Segment中是否有相同Key存在，如果已经存在，那么根据onlyIfAbsent参数确定是否替换(如果为true,不替换，如果为false,替换掉value),然后返回替换的value,如果不存在，那么新生成一个HashEntry,并且根据一开始计算出来的index放到数组指定位置，并且累积元素计数，返回put的值。最后unlock()释放掉锁.

4. CopyOnWriteArrayList和CopyOnWriteArraySet

             CopyOnWriteList是线程安全的List实现，其底层数据存储结构为数组(Object[] array),它在读操作远远多于写操作的场景下表现良好，这其中的原因在于其读操作(get(),indexOf(),isEmpty(),contains())不加任何锁，而写操作(set(),add(),remove())通过Arrays.copyOf()操作拷贝当前底层数据结构(array)，在其上面做完增删改等操作，再将新的数组置为底层数据结构，同时为了避免并发增删改， CopyOnWriteList在这些写操作上通过一个ReetranLock进行并发控制。另外需要注意的是，CopyOnWriteList所实现的迭代器其数据也是底层数组镜像，所以在CopyOnWriteList进行interator，同时并发增删改CopyOnWriteList里的数据实不会抛“ConcurrentModificationException”，当然在迭代器上做remove,add,set也是无效的（抛UnsupportedOperationExcetion），因为迭代器上的数据只是当前List的数据数组的一个拷贝而已。

            CopyOnWriteSet是一个线程安全的Set实现，然后持有一个CopyOnWriteList实例,其所有的操作都是这个CopyOnWriteList实例来实现的。CopyOnWriteSet与CopyOnWriteList的区别实际上就是Set与List的区别，前者不允许有重复的元素，后者是可以的，所以CopyOnWriteSet的add和addAll两个操作使用的是其内部CopyOnWriteList实例的addAbsent()和addAllAbsent()两个防止重复元素的方法，addAbsent()实现是拷贝底层数据数组，然后逐一比较是否相同，如果有一个相同，那么直接返回false,说明插入失败，如果和其他元素不同，那么将元素加入到新的数组中，最后置回新的数组, addAllAbsent()方法实现则是能有多少数据插入就插入，也就是说addAllAbsent一个集合的数据，可能只有一部分插入成功，另外一部分因为元素相同而遭丢弃，完成后返回插入的元素。

 

 

           Java库本身就有多种线程安全的容器和同步工具，其中同步容器包括两部分：一个是Vector和Hashtable。另外还有JDK1.2中加入的同步包装类，这些类都是由Collections.synchronizedXXX工厂方法。同步容器都是线程安全的，但是对于复合操作，缺有些缺点：
①　迭代：在查觉到容器在迭代开始以后被修改，会抛出一个未检查异常ConcurrentModificationException，为了避免这个异常，需要在迭代期间，持有一个容器锁。但是锁的缺点也很明显，就是对性能的影响。
②　隐藏迭代器：StringBuilder的toString方法会通过迭代容器中的每个元素，另外容器的hashCode和equals方法也会间接地调用迭代。类似地，contailAll、removeAll、retainAll方法，以及容器作为参数的构造函数，都会对容器进行迭代。
③　缺少即加入等一些复合操作



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

}

getLast和deleteLast都是复合操作，由先前对原子性的分析可以判断，这依然存在线程安全问题，有可能会抛出ArrayIndexOutOfBoundsException的异常，错误产生的逻辑如下所示：


解决办法就是通过对这些复合操作加锁
1 并发容器类
        正是由于同步容器类有以上问题，导致这些类成了鸡肋，于是Java 5推出了并发容器类，Map对应的有ConcurrentHashMap，List对应的有CopyOnWriteArrayList。与同步容器类相比，它有以下特性：
1.1 ConcurrentHashMap
· 更加细化的锁机制。同步容器直接把容器对象做为锁，这样就把所有操作串行化，其实这是没必要的，过于悲观，而并发容器采用更细粒度的锁机制，名叫分离锁，保证一些不会发生并发问题的操作进行并行执行

· 附加了一些原子性的复合操作。比如putIfAbsent方法

· 迭代器的弱一致性，而非“及时失败”。它在迭代过程中不再抛出Concurrentmodificationexception异常，而是弱一致性。

· 在并发高的情况下，有可能size和isEmpty方法不准确，但真正在并发环境下这些方法也没什么作用。

· 另外，它还有一些附加的原子操作，缺少即加入、相等便移除、相等便替换。
putIfAbsent(K key, V value)，缺少即加入（如果该键已经存在，则不加入）
          如果指定键已经不再与某个值相关联，则将它与给定值关联。
类似于下面的操作
If(!map.containsKey(key)){
return map.put(key,value);
}else{
return map.get(key);
}

remove(Object key, Object value),相等便移除
          只有目前将键的条目映射到给定值时，才移除该键的条目。
类似于下面的：
if(map.containsKey(key) && map.get(key).equals(value)){

Map.remove();

return true;

}else{

return false;

}


replace(K key, V value)

replace(K key, V oldValue, V newValue),相等便替换。
          只有目前将键的条目映射到某一值时，才替换该键的条目。

上面提到的三个，都是原子的。在一些缓存应用中可以考虑代替HashMap/Hashtable。
1.2 CopyOnWriteArrayList和CopyOnWriteArraySet
· CopyOnWriteArrayList采用写入时复制的方式避开并发问题。这其实是通过冗余和不可变性来解决并发问题，在性能上会有比较大的代价，但如果写入的操作远远小于迭代和读操作，那么性能就差别不大了。

          应用它们的场合通常在数组相对较小，并且遍历操作的数量大大超过可变操作的数量时，这种场合应用它们非常好。它们所有可变的操作都是先取得后台数组的副本，对副本进行更改，然后替换副本，这样可以保证永远不会抛出ConcurrentModificationException移除。

2 队列
          Java中的队列接口就是Queue，它有会抛出异常的add、remove方法，在队尾插入元素以及对头移除元素，还有不会抛出异常的，对应的offer、poll方法。

2.1 LinkedList
           List实现了deque接口以及List接口，可以将它看做是这两种的任何一种。
Queue queue=new LinkedList();
queue.offer("testone");
queue.offer("testtwo");
queue.offer("testthree");
queue.offer("testfour");
System.out.println(queue.poll()); //testone

2.2 PriorityQueue
        一个基于优先级堆（简单的使用链表的话，可能插入的效率会比较低O(N)）的无界优先级队列。优先级队列的元素按照其自然顺序进行排序，或者根据构造队列时提供的 Comparator 进行排序，具体取决于所使用的构造方法。优先级队列不允许使用 null 元素。依靠自然顺序的优先级队列还不允许插入不可比较的对象。
queue=new PriorityQueue();
queue.offer("testone");
queue.offer("testtwo");
queue.offer("testthree");
queue.offer("testfour");
System.out.println(queue.poll()); //testfour



2.3 ConcurrentLinkedQueue
         基于链节点的，线程安全的队列。并发访问不需要同步。在队列的尾部添加元素，并在头部删除他们。所以只要不需要知道队列的大小，并发队列就是比较好的选择。

3 阻塞队列
3.1 生产者和消费者模式
         生产者和消费者模式，生产者不需要知道消费者的身份或者数量，甚至根本没有消费者，他们只负责把数据放入队列。类似地，消费者也不需要知道生产者是谁，以及是谁给他们安排的工作。
           而Java知道大家清楚这个模式的并发复杂性，于是乎提供了阻塞队列（BlockingQueue）来满足这个模式的需求。阻塞队列说起来很简单，就是当队满的时候写线程会等待，直到队列不满的时候；当队空的时候读线程会等待，直到队不空的时候。实现这种模式的方法很多，其区别也就在于谁的消耗更低和等待的策略更优。以LinkedBlockingQueue的具体实现为例，它的put源码如下：

public void put(E e) throws InterruptedException {

        if (e == null) throw new NullPointerException();

        int c = -1;

        final ReentrantLock putLock = this.putLock;

        final AtomicInteger count = this.count;

        putLock.lockInterruptibly();

        try {

           

            try {

                while (count.get() == capacity)

                    notFull.await();

            } catch (InterruptedException ie) {

                notFull.signal();

// propagate to a non-interrupted thread

                throw ie;

            }

            insert(e);

            c = count.getAndIncrement();

            if (c + 1 < capacity)

                notFull.signal();

        } finally {

            putLock.unlock();

        }

        if (c == 0)

            signalNotEmpty();

}
        撇开其锁的具体实现，其流程就是我们在操作系统课上学习到的标准生产者模式，看来那些枯燥的理论还是有用武之地的。其中，最核心的还是Java的锁实现，有兴趣的朋友可以再进一步深究一下。
         阻塞队列Blocking queue，提供了可阻塞的put和take方法，他们与可定时的offer和poll方法是等价。Put方法简化了处理，如果是有界队列，那么当队列满的时候，生成者就会阻塞，从而改消费者更多的追赶速度。 
3.2 ArrayBlockingQueue和LinkedBlockingQueue
          FIFO的队列，与LinkedList(由链节点支持，无界)和ArrayList（由数组支持，有界）相似（Linked有更好的插入和移除性能，Array有更好的查找性能，考虑到阻塞队列的特性，移除头部，加入尾部，两个都区别不大），但是却拥有比同步List更好的并发性能。
另外，LinkedList永远不会等待，因为他是无界的。

BlockingQueue<String> queue=new ArrayBlockingQueue<String>(5);
Producer p=new Producer(queue);
Consumer c1=new Consumer(queue);
Consumer c2=new Consumer(queue);

new Thread(p).start();
new Thread(c1).start();
new Thread(c2).start();

class Producer implements Runnable {

   private final BlockingQueue queue; 
   Producer(BlockingQueue q) { queue = q; }   

   public void run() {

     try {

      for(int i=0;i<100;i++){

      queue.put(produce());

      }

     

     } catch (InterruptedException ex) {}

   }

  

   String produce() {

   String temp=""+(char)('A'+(int)(Math.random()*26));

   System.out.println("produce"+temp);

   return temp;

   }

}





class Consumer implements Runnable {

   private final BlockingQueue queue;

   Consumer(BlockingQueue q) { queue = q; }

   public void run() {

     try {

      for(int i=0;i<100;i++){

      consume(queue.take());

      }

     } catch (InterruptedException ex) {}

   }

   void consume(Object x) {

   System.out.println("cousume"+x.toString());

   }

}



输出：

produceK

cousumeK

produceV

cousumeV

produceQ

cousumeQ

produceI

produceD

produceI

produceG

produceA

produceE

cousumeD

3.3 PriorityBlockingQueue
           一个按优先级堆支持的无界优先级队列队列，如果不希望按照FIFO的顺序进行处理，它非常有用。它可以比较元素本身的自然顺序，也可以使用一个Comparator排序。

3.4 DelayQueue
           一个优先级堆支持的，基于时间的调度队列。加入到队列中的元素必须实现新的Delayed接口（只有一个方法，Long getDelay(java.util.concurrent.TimeUnit unit)），添加可以理立即返回，但是在延迟时间过去之前，不能从队列中取出元素，如果多个元素的延迟时间已到，那么最早失效链接/失效时间最长的元素将第一个取出。
static class NanoDelay implements Delayed{

long tigger;

NanoDelay(long i){

tigger=System.nanoTime()+i;

}

public boolean equals(Object other){

return ((NanoDelay)other).tigger==tigger;

}

public long getDelay(TimeUnit unit) {

long n=tigger-System.nanoTime();

return unit.convert(n, TimeUnit.NANOSECONDS);

}
public long getTriggerTime(){

return tigger;

}
public int compareTo(Delayed o) {

long i=tigger;

long j=((NanoDelay)o).tigger;

if(i<j){

return -1;

}

if(i>j)

return 1;

return 0;

}

}
public static void main(String[] args) throws InterruptedException{

Random random=new Random();

DelayQueue<NanoDelay> queue=new DelayQueue<NanoDelay>();

for(int i=0;i<5;i++){

queue.add(new NanoDelay(random.nextInt(1000)));
}

long last=0;

for(int i=0;i<5;i++){

NanoDelay delay=(NanoDelay)(queue.take());

long tt=delay.getTriggerTime();

System.out.println("Trigger time："+tt);
if(i!=0){

System.out.println("Data: "+(tt-last));

}

last=tt;

}

}

3.5 SynchronousQueue
           不是一个真正的队列，因为它不会为队列元素维护任何存储空间，不过它维护一个排队的线程清单，这些线程等待把元素加入（enqueue）队列或者移出（dequeue）队列。也就是说，它非常直接的移交工作，减少了生产者和消费者之间移动数据的延迟时间，另外，也可以更快的知道反馈信息，当移交被接受时，它就知道消费者已经得到了任务。

          因为SynChronousQueue没有存储的能力，所以除非另一个线程已经做好准备，否则put和take会一直阻止。它只有在消费者比较充足的时候比较合适。

4 双端队列（Deque）
            JAVA6中新增了两个容器Deque和BlockingDeque，他们分别扩展了Queue和BlockingQueue。Deque它是一个双端队列，允许高效的在头和尾分别进行插入和删除，它的实现分别是ArrayDeque和LinkedBlockingQueue。

双端队列使得他们能够工作在一种称为“窃取工作”的模式上面。

5 最佳实践
1..同步的（synchronized）+HashMap，如果不存在，则计算，然后加入，该方法需要同步。
HashMap cache=new HashMap();
public synchronized V compute(A arg){

V result=cace.get(arg);

Ii(result==null){

result=c.compute(arg);

Cache.put(result);

}

Return result;

}





2.用ConcurrentHashMap代替HashMap+同步.，这样的在get和set的时候也基本能保证原子性。但是会带来重复计算的问题.

Map<A,V> cache=new ConcurrentHashMap<A,V>();

public  V compute(A arg){

V result=cace.get(arg);

Ii(result==null){

result=c.compute(arg);

Cache.put(result);

}

Return result;

}

3.采用FutureTask代替直接存储值，这样可以在一开始创建的时候就将Task加入

Map<A,FutureTask<V>> cache=new ConcurrentHashMap<A,FutureTask<V>>();

public  V compute(A arg){

FutureTask <T> f=cace.get(arg);

//检查再运行的缺陷

Ii(f==null){

Callable<V> evel=new Callable(){

Public V call() throws ..{

return c.compute(arg);

}

};

FutureTask <T> ft=new FutureTask<T>(evel);

f=ft;

cache.put(arg,ft;

ft.run();

}

Try{

//阻塞，直到完成

return f.get();

}cach(){

}

}
4.上面还有检查再运行的缺陷，在高并发的情况下啊，双方都没发现FutureTask，并且都放入Map（后一个被前一个替代），都开始了计算。
这里的解决方案在于，当他们都要放入Map的时候，如果可以有原子方法，那么已经有了以后，后一个FutureTask就加入，并且启动。

public  V compute(A arg){

FutureTask <T> f=cace.get(arg);

//检查再运行的缺陷

Ii(f==null){

Callable<V> evel=new Callable(){

Public V call() throws ..{

return c.compute(arg);

}

};

FutureTask <T> ft=new FutureTask<T>(evel);

f=cache.putIfAbsent(args,ft); //如果已经存在，返回存在的值，否则返回null

if(f==null){f=ft;ft.run();} //以前不存在，说明应该开始这个计算

else{ft=null;} //取消该计算

}

Try{

//阻塞，直到完成

return f.get();

}cach(){

}

}



5.上面的程序上来看已经完美了，不过可能带来缓存污染的可能性。如果一个计算被取消或者失败，那么这个键以后的值永远都是失败了；一种解决方案是，发现取消或者失败的task，就移除它，如果有Exception，也移除。

6.另外，如果考虑缓存过期的问题，可以为每个结果关联一个过去时间，并周期性的扫描，清除过期的缓存。（过期时间可以用Delayed接口实现，参考DelayQueue，给他一个大于当前时间XXX的时间，，并且不断减去当前时间，直到返回负数，说明延迟时间已到了。）

Java集合容器总结。
按数据结构主要有以下几类：
1,内置容器:数组
2,list容器:

           Vetor,Stack，ArrayList,LinkedList, CopyOnWriteArrayList(1.5),AttributeList(1.5),RoleList(1.5),RoleUnresolvedList(1.5)，
ConcurrentLinkedQueue(1.5),ArrayBlockingQueue(1.5)，LinkedBlockingQueue（1.5）， PriorityQueue（1.5），PriorityBlockingQueue（1.5），SynchronousQueue（1.5）
3,set容器：

            HashSet（1.2）,LinkedHashSet（1.4）,TreeSet（1.2），
CopyOnWriteArraySet（1.5），EnumSet（1.5），JobStateReasons。
4,map容器：

         Hashtable,HashMap（1.2）,TreeMap（1.2），LinkedHashMap（1.4），WeakHashMap（1.2）， IdentityHashMap（1.4），ConcurrentMap（1.5），concurrentHashMap（1.5）。
     

         Set 接口继承 Collection，但不允许重复，使用自己内部的一个排列机制。 List 接口继承 Collection，允许重复，以元素安插的次序来放置元素，不会重新排列。 
           Map接口是一组成对的键－值对象，即所持有的是key-value pairs。Map中不能有重复的key。拥有自己的内部排列机制。

按新旧主要有以下几类：
Java1.2前的容器：Vector,Stack,Hashtable。
Java1.2的容器：HashSet，TreeSet，HashMap，TreeMap，WeakHashMap
Java1.4的容器：LinkedHashSet，LinkedHashMap，IdentityHashMap，ConcurrentMap，concurrentHashMap
java1.5新增：    CopyOnWriteArrayList,AttributeList,RoleList,RoleUnresolvedList，
ConcurrentLinkedQueue,ArrayBlockingQueue，LinkedBlockingQueue，PriorityBlockingQueue
ArrayBlockingQueue，CopyOnWriteArraySet，EnumSet，
未知：JobStateReasons
按线程安全主要有以下几类：
线程安全
一，使用锁：
  完全不支持并发：
  list容器：Vetor，Stack，CopyOnWriteArrayList，ArrayBlockingQueue，
  LinkedBlockingQueue，PriorityBlockingQueue，SynchronousQueue
  set容器：CopyOnWriteArraySet
  map容器：Hashtable
  部分支持并发：
  list容器：无
  set容器：无
  map容器：concurrentHashMap

使用非阻塞算法：
list容器：ConcurrentLinkedQueue
set容器：无
map容器：无
二，非线程安全：
         list容器：ArrayList,LinkedList，AttributeList，RoleList,RoleUnresolvedList，PriorityQueue
set容器：HashSet，TreeSet，LinkedHashSet，EnumSet
map容器：HashMap，TreeMap，LinkedHashMap，WeakHashMap，IdentityHashMap,EnumMap

按遍历安全主要有以下几类：
一，遍历安全：
可并发遍历：
   list容器：CopyOnWriteArrayList，ConcurrentLinkedQueue
   set容器：CopyOnWriteArraySet，EnumSet，EnumMap
   map容器：无
不可并发遍历：
   list容器：Vetor，Stack，Hashtable，ArrayBlockingQueue，
   LinkedBlockingQueue，PriorityBlockingQueue，SynchronousQueue
   set容器：无
   map容器：Hashtable，concurrentHashMap
注意1：concurrentHashMap迭代器它们不会抛出ConcurrentModificationException。不过，迭代器被设计成每次仅由一个线程使用。
二，遍历不安全：
会抛异常ConcurrentModificationException：
list容器：ArrayList，LinkedList，AttributeList，RoleList，RoleUnresolvedList
set容器：HashSet，TreeSet，TreeSet，LinkedHashSet
map容器：HashMap，TreeMap，LinkedHashMap，WeakHashMap，IdentityHashMap
注意1：返回的迭代器是弱一致 的：它们不会抛出 ConcurrentModificationException，
   也不一定显示在迭代进行时发生的任何映射修改的效果的容器有：
   EnumSet,EnumMap

按遍历是否有序性分类
存储数据有序：
list容器:

         ConcurrentLinkedQueue(1.5),ArrayBlockingQueue(1.5)，LinkedBlockingQueue（1.5），
SynchronousQueue（1.5）
set容器：

   TreeSet（1.2）.（他们实现了set接口），
CopyOnWriteArraySet（1.5），EnumSet（1.5），JobStateReasons。
map容器：TreeMap（1.2），LinkedHashMap（1.4） 。

 
一定规则下存储数据有序：
list容器:

        Stack，Vetor，ArrayList，LinkedList，CopyOnWriteArrayList(1.5),AttributeList(1.5),RoleList(1.5),RoleUnresolvedList(1.5)
set容器：无
map容器：无

遍历无序但移除有序：
list容器:PriorityQueue（1.5），PriorityBlockingQueue（1.5）
set容器：无
map容器：无
无论如何都无序：
list容器:无
set容器：HashSet（1.2）,LinkedHashSet（1.4）
map容器：Hashtable,HashMap（1.2），WeakHashMap（1.2），IdentityHashMap（1.4），
ConcurrentMap（1.5），concurrentHashMap（1.5）

可以按自然顺序（参见 Comparable）或比较器进行排序的有：
list容器：PriorityQueue（1.5），PriorityBlockingQueue
set容器：TreeSet（1.2）
map容器：TreeMap（1.2）
实现了RandomAccess接口的有：
ArrayList, AttributeList, CopyOnWriteArrayList, RoleList, RoleUnresolvedList, Stack, Vector
         RandomAccess接口是List 实现所使用的标记接口，用来表明其支持快速（通常是固定时间）随机访问。此接口的主要目的是允许一般的算法更改其行为，从而在将其应用到随机或连续访问列表时能提供良好的性能。
           在对List特别的遍历算法中，要尽量来判断是属于 RandomAccess(如ArrayList)还是SequenceAccess(如LinkedList），
             因为适合RandomAccess List的遍历算法，用在SequenceAccess List上就差别很大，
即对于实现了RandomAccess接口的类实例而言，此循环
for (int i=0, i<list.size(); i++)
   list.get(i);
的运行速度要快于以下循环：
     for (Iterator i=list.iterator(); i.hasNext(); )
    i.next();