Go语言视频零基础入门到精通

javalearner

这是Ted Neward在IBM developerWorks中5 things系列文章中的一篇，讲述了关于java并发集合API的一些应用窍门，值得大家学习。(2010.05.24最后更新)

     摘要：编写既要性能良好又要防止应用崩溃的多线程代码确实很难--这也正是我们需要java.util.concurrent的原因。Ted Neward向你展示了像CopyOnWriteArrayList，BlockingQueue和ConcurrentMap这样的并发集合类是如何为了并发编程需要而改进标准集合类的。

     并发集合API是Java 5的一大新特性，但由于对Annotation和泛型的热捧，许多Java开发者忽视了这些API。另外(可能更真实的是)，因为许多开发者猜想并发集合 API肯定很复杂，就像去尝试解决一些问题那样，所以开发者们会回避java.util.concurrent包。
     事实上，java.util.concurrent的很多类并不需要你费很大力就能高效地解决通常的并发问题。继续看下去，你就能学到 java.util.concurrent中的类，如CopyOnWriteArrayList和BlockingQueue，是怎样帮助你解决多线程编程可怕的挑战。  
  
   
   
     
   

     
   
  



  

  

  1. TimeUnit
     java.util.concurrent.TimeUnit本身并不是集合框架类，这个枚举使得代码非常易读。使用TimeUnit能够将开发者从与毫秒相关的困苦中解脱出来，转而他们自己的方法或API。
     TimeUnit能与所有的时间单元协作，范围从毫秒和微秒到天和小时，这就意味着它能处理开发者可能用到的几乎所有时间类型。还要感谢这个枚举类型声明的时间转换方法，当时间加快时，它甚至能细致到把小时转换回毫秒。

2. CopyOnWriteArrayList
     制作数组的干净复本是一项成本极高的操作，在时间和内存这两方面均有开销，以至于在通常的应用中不能考虑该方法；开发者常常求助于使用同步的 ArrayList来替代前述方法。但这也是一个比较有代价的选项，因为当每次你遍历访问该集合中的内容时，你不得不同步所有的方法，包括读和写，以确保内存一致性。
     在有大量用户在读取ArrayList而只有很少用户对其进行修改的这一场景中，上述方法将使成本结构变得缓慢。
     CopyOnWriteArrayList就是解决这一问题的一个极好的宝贝工具。它的Javadoc描述到，ArrayList通过创建数组的干净复本来实现可变操作(添加，修改，等等)，而CopyOnWriteArrayList则是ArrayList的一个"线程安全"的变体。
     对于任何修改操作，该集合类会在内部将其内容复制到一个新数组中，所以当读用户访问数组的内容时不会招致任何同步开销(因为它们没有对可变数据进行操作)。
     本质上，创建CopyOnWriteArrayList的想法，是出于应对当ArrayList无法满足我们要求时的场景：经常读，而很少写的集合对象，例如针对JavaBean事件的Listener。

3. BlockingQueue
     BlockingQueue接口表明它是一个Queue，这就意味着它的元素是按先进先出(FIFO)的次序进行存储的。以特定次序插入的元素会以相同的次序被取出--但根据插入保证，任何从空队列中取出元素的尝试都会堵塞调用线程直到该元素可被取出时为止。同样地，任何向一个已满队列中插入元素的尝试将会堵塞调用线程直到该队列的存储空间有空余时为止。
     在不需要显式地关注同步问题时，如何将由一个线程聚集的元素"交给"另一个线程进行处理呢，BlockingQueue很灵巧地解决了这个问题。Java Tutorial中Guarded Blocks一节是很好的例子。它使用手工同步和wait()/notifyAll()方法创建了一个单点(single-slot)受限缓冲，当一个新的元素可被消费且当该点已经准备好被一个新的元素填充时，该方法就会在线程之间发出信号。(详情请见Guarded Blocks)
     尽管教程Guarded Blocks中的代码可以正常工作，但它比较长，有些凌乱，而且完全不直观。诚然，在Java平台的早期时代，Java开发者们不得不；但现在已经是 2010年了--问题已经得到改进？
     清单1展示的程序重写了Guarded Blocks中的代码，其中我使用ArrayBlockingQueue替代了手工编写的Drop。

清单1. BlockingQueue
  
   
   
   import
    java.util.
   *
   ;

   import
    java.util.concurrent.
   *
   ;


   class
    Producer
     
   implements
    Runnable
{
     
   private
    BlockingQueue
   <
   String
   >
    drop;
     List
   <
   String
   >
    messages
   =
    Arrays.asList(
         
   "
   Mares eat oats
   "
   ,
         
   "
   Does eat oats
   "
   ,
         
   "
   Little lambs eat ivy
   "
   ,
         
   "
   Wouldn"t you eat ivy too?
   "
   );
         
     
   public
    Producer(BlockingQueue
   <
   String
   >
    d) {
   this
   .drop
   =
    d; }
     
     
   public
   
   void
    run()
     {
         
   try
   
         {
            
   for
    (String s : messages)
                 drop.put(s);
             drop.put(
   "
   DONE
   "
   );
         }
         
   catch
    (InterruptedException intEx)
         {
             System.out.println(
   "
   Interrupted!
   "
   
   +
   
                 
   "
   Last one out, turn out the lights!
   "
   );
         }
     }   
}


   class
    Consumer
     
   implements
    Runnable
{
     
   private
    BlockingQueue
   <
   String
   >
    drop;
     
   public
    Consumer(BlockingQueue
   <
   String
   >
    d) {
   this
   .drop
   =
    d; }
     
     
   public
   
   void
    run()
     {
         
   try
   
         {
             String msg
   =
   
   null
   ;
            
   while
    (
   !
   ((msg
   =
    drop.take()).equals(
   "
   DONE
   "
   )))
                 System.out.println(msg);
         }
         
   catch
    (InterruptedException intEx)
         {
             System.out.println(
   "
   Interrupted!
   "
   
   +
   
                 
   "
   Last one out, turn out the lights!
   "
   );
         }
     }
}


   public
   
   class
    ABQApp
{
     
   public
   
   static
   
   void
    main(String[] args)
     {
         BlockingQueue
   <
   String
   >
    drop
   =
   
   new
    ArrayBlockingQueue(
   1
   ,
   true
   );
         (
   new
    Thread(
   new
    Producer(drop))).start();
         (
   new
    Thread(
   new
    Consumer(drop))).start();
     }
}
   
  

  ArrayBlockingQueue也崇尚"公平"--即意味着，它能给予读和写线程先进先出的访问次序。该方法可能是一种更高效的策略，但它也加大了造成线程饥饿的风险。(就是说，当其它读线程持有锁时，该策略可更高效地允许读线程进行执行，但这也就会产生读线程的常量流使写线程总是无法执行的风险)
     BlockingQueue也支持在方法中使用时间参数，当插入或取出元素出了问题时，方法需要返回以发出操作失败的信号，而该时间参数指定了在返回前应该阻塞多长时间。

4. ConcurrentMap
     Map有一些细微的并发Bug，会使许多粗心的Java开发者误入歧途。ConcurrentMap则是一个简单的决定方案。
     当有多个线程在访问一个Map时，通常在储存一个键/值对之前通常会使用方法containsKey()或get()去确定给出的键是否存在。即使用同步的Map，某个线程仍可在处理的过程中潜入其中，然后获得对Map的控制权。问题在于，在get()方法的开始处获得了锁，然后在调用方法put()去重新获得该锁之前会先释放它。这就导致了竞争条件：两个线程之间的竞争，根据哪个线程先执行，其结果将不尽相同。
     如果两个线程在同一时刻调用一个方法，一个测试键是否存在，另一个则置入新的键/值对，那么在此过程中，第一个线程的值将会丢失。幸运地是，ConcurrentMap接口支持一组额外的方法，设计这些方法是为了在一个锁中做两件事情：例如，putIfAbsent()首先进行测试，之后只有当该键还未存储到Map中时，才执行置入操作。

5. SynchronousQueues
     根据Javadoc的描述，SynchronousQueue是一个很有趣的创造物：
     一个阻塞队列在每次的插入操作中必须等等另一线程执行对应的删除线程，反之亦然。同步队列并没有任何内部的存储空间，一个都没有。
     本质上，SynchronousQueue是之前提及的BlockingQueue的另一种实现。使用ArrayBlockingQueue利用的阻塞语义，SynchronousQueue给予我们一种极轻量级的途径在两个线程之间交换单个元素。在清单2中，我用SynchronousQueue替代 ArrayBlockingQueue重写了清单1的代码：

清单2 SynchronousQueue
  
   
   
   import
    java.util.
   *
   ;

   import
    java.util.concurrent.
   *
   ;


   class
    Producer
     
   implements
    Runnable
{
     
   private
    BlockingQueue
   <
   String
   >
    drop;
     List
   <
   String
   >
    messages
   =
    Arrays.asList(
         
   "
   Mares eat oats
   "
   ,
         
   "
   Does eat oats
   "
   ,
         
   "
   Little lambs eat ivy
   "
   ,
         
   "
   Wouldn"t you eat ivy too?
   "
   );
         
     
   public
    Producer(BlockingQueue
   <
   String
   >
    d) {
   this
   .drop
   =
    d; }
     
     
   public
   
   void
    run()
     {
         
   try
   
         {
            
   for
    (String s : messages)
                 drop.put(s);
             drop.put(
   "
   DONE
   "
   );
         }
         
   catch
    (InterruptedException intEx)
         {
             System.out.println(
   "
   Interrupted!
   "
   
   +
   
                 
   "
   Last one out, turn out the lights!
   "
   );
         }
     }   
}


   class
    Consumer
     
   implements
    Runnable
{
     
   private
    BlockingQueue
   <
   String
   >
    drop;
     
   public
    Consumer(BlockingQueue
   <
   String
   >
    d) {
   this
   .drop
   =
    d; }
     
     
   public
   
   void
    run()
     {
         
   try
   
         {
             String msg
   =
   
   null
   ;
            
   while
    (
   !
   ((msg
   =
    drop.take()).equals(
   "
   DONE
   "
   )))
                 System.out.println(msg);
         }
         
   catch
    (InterruptedException intEx)
         {
             System.out.println(
   "
   Interrupted!
   "
   
   +
   
                 
   "
   Last one out, turn out the lights!
   "
   );
         }
     }
}


   public
   
   class
    SynQApp
{
     
   public
   
   static
   
   void
    main(String[] args)
     {
         BlockingQueue
   <
   String
   >
    drop
   =
   
   new
    SynchronousQueue
   <
   String
   >
   ();
         (
   new
    Thread(
   new
    Producer(drop))).start();
         (
   new
    Thread(
   new
    Consumer(drop))).start();
     }
}
   
  
上述实现看起来几乎相同，但该应用程序已新加了一个好处，在这个实现中，只有当有线程正在等待消费某个元素时，SynchronousQueue才会允许将该元素插入到队列中。
就实践方式来看，SynchronousQueue类似于Ada或CSP等语言中的"交会通道(Rendezvous Channel)"。在其它环境中，有时候被称为"连接"。

结论
     当Java运行时类库预先已经提供了方便使用的等价物时，为什么还要费力地向集合框架中引入并发呢？本系列的下一篇文章将探索 java.util.concurrent命名空间的更多内容。
  

复制代码