多线程编程的常用类（CountDownLatch, Semaphore, CyclicBarrier 和 Exchanger）

Myth丶恋晨 2022-08-19 15:21 195阅读 0赞

（1）CountDownLatch

![复制代码][copycode.gif]

/**

CountDownLatch类是一个同步计数器,构造时传入int参数,该参数就是计数器的初始值，每调用一次countDown()方法，计数器减1,计数器大于0 时，await()方法会阻塞程序继续执行

CountDownLatch如其所写，是一个倒计数的锁存器，当计数减至0时触发特定的事件。利用这种特性，可以让主线程等待子线程的结束。下面以一个模拟运动员比赛的例子加以说明。

import java.util.concurrent.CountDownLatch;
      import java.util.concurrent.Executor;
      import java.util.concurrent.ExecutorService;
      import java.util.concurrent.Executors;
     
      public class CountDownLatchDemo {
         private static final int PLAYER_AMOUNT = 5;
         public CountDownLatchDemo() {
             // TODO Auto-generated constructor stub    
          }
         /**
          * @param args
          */
         public static void main(String[] args) {
             // TODO Auto-generated method stub
             //对于每位运动员，CountDownLatch减1后即结束比赛
             CountDownLatch begin = new CountDownLatch(1);
             //对于整个比赛，所有运动员结束后才算结束
             CountDownLatch end = new CountDownLatch(PLAYER_AMOUNT);
             Player[] plays = new Player[PLAYER_AMOUNT];
             
             for(int i=0;i<PLAYER_AMOUNT;i++)
                 plays[i] = new Player(i+1,begin,end);
             
             //设置特定的线程池，大小为5
             ExecutorService exe = Executors.newFixedThreadPool(PLAYER_AMOUNT);
             for(Player p:plays)
                 exe.execute(p);            //分配线程
             System.out.println("Race begins!");
             begin.countDown();
             try{
                 end.await();            //等待end状态变为0，即为比赛结束
             }catch (InterruptedException e) {
                 // TODO: handle exception
                 e.printStackTrace();
             }finally{
                 System.out.println("Race ends!");
             }
             exe.shutdown();
         }
     }

![复制代码][copycode.gif]

接下来是Player类

![复制代码][copycode.gif]

import java.util.concurrent.CountDownLatch;
     
     
     public class Player implements Runnable {
     
         private int id;
         private CountDownLatch begin;
         private CountDownLatch end;
         public Player(int i, CountDownLatch begin, CountDownLatch end) {
             // TODO Auto-generated constructor stub
             super();
             this.id = i;
             this.begin = begin;
             this.end = end;
         }
     
         @Override
         public void run() {
             // TODO Auto-generated method stub
             try{
                 begin.await();        //等待begin的状态为0
                 Thread.sleep((long)(Math.random()*100));    //随机分配时间，即运动员完成时间
                 System.out.println("Play"+id+" arrived.");
             }catch (InterruptedException e) {
                 // TODO: handle exception
                 e.printStackTrace();
             }finally{
                 end.countDown();    //使end状态减1，最终减至0
             }
         }
     }

![复制代码][copycode.gif]  
（2）Semaphore  
  
Semaphore也是一个线程同步的辅助类，可以维护当前访问自身的线程个数，并提供了同步机制。使用Semaphore可以控制同时访问资源的线程个数，例如，实现一个文件允许的并发访问数。

**Semaphore的主要方法摘要：**

void acquire()`:从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。`

void release():释放一个许可，将其返回给信号量。

int availablePermits():返回此信号量中当前可用的许可数。

boolean hasQueuedThreads():查询是否有线程正在等待获取。

下面是一个例子：

![复制代码][copycode.gif]

1 package com.thread;
     2 
     3 import java.util.concurrent.ExecutorService;
     4 import java.util.concurrent.Executors;
     5 import java.util.concurrent.Semaphore;
     6 
     7 public class SemaphoreTest {
     8     public static void main(String[] args) {
     9         ExecutorService service = Executors.newCachedThreadPool();
    10         final  Semaphore sp = new Semaphore(3);//创建Semaphore信号量，初始化许可大小为3
    11         for(int i=0;i<10;i++){
    12             try {
    13                 Thread.sleep(100);
    14             } catch (InterruptedException e2) {
    15                 e2.printStackTrace();
    16             }
    17             Runnable runnable = new Runnable(){
    18                     public void run(){
    19                     try {
    20                         sp.acquire();//请求获得许可，如果有可获得的许可则继续往下执行，许可数减1。否则进入阻塞状态
    21                     } catch (InterruptedException e1) {
    22                         e1.printStackTrace();
    23                     }
    24                     System.out.println("线程" + Thread.currentThread().getName() + 
    25                             "进入，当前已有" + (3-sp.availablePermits()) + "个并发");
    26                     try {
    27                         Thread.sleep((long)(Math.random()*10000));
    28                     } catch (InterruptedException e) {
    29                         e.printStackTrace();
    30                     }
    31                     System.out.println("线程" + Thread.currentThread().getName() + 
    32                             "即将离开");                    
    33                     sp.release();//释放许可，许可数加1
    34                     //下面代码有时候执行不准确，因为其没有和上面的代码合成原子单元
    35                     System.out.println("线程" + Thread.currentThread().getName() + 
    36                             "已离开，当前已有" + (3-sp.availablePermits()) + "个并发");                    
    37                 }
    38             };
    39             service.execute(runnable);            
    40         }
    41     }
    42 
    43 }

![复制代码][copycode.gif]

单个信号量的Semaphore对象可以实现互斥锁的功能，并且可以是由一个线程获得了“锁”，再由另一个线程释放“锁”，这可应用于死锁恢复的一些场合。

![复制代码][copycode.gif]

1 package com.thread;
     2 import java.util.concurrent.ExecutorService;
     3 import java.util.concurrent.Executors;
     4 import java.util.concurrent.Semaphore;
     5 import java.util.concurrent.locks.Lock;
     6 import java.util.concurrent.locks.ReentrantLock;
     7 
     8 public class LockTest {
     9     public static void main(String[] args) {
    10         final Business business = new Business();
    11         ExecutorService executor =  Executors.newFixedThreadPool(3);
    12         for(int i=0;i<3;i++)
    13         {
    14             executor.execute(
    15                     new Runnable()
    16                     {
    17                         public void run()
    18                         {
    19                             business.service();
    20                         }
    21                     }
    22             
    23             );
    24         }
    25         executor.shutdown();
    26     }
    27     
    28     private static class Business
    29     {
    30         private int count;
    31         Lock lock = new ReentrantLock();
    32         Semaphore sp = new Semaphore(1);
    33         public void service() 
    34         {
    35             //lock.lock();
    36             try {
    37                 sp.acquire(); //当前线程使用count变量的时候将其锁住，不允许其他线程访问
    38             } catch (InterruptedException e1) {
    39                 e1.printStackTrace();
    40             }
    41             try {
    42                 count++;
    43                 try {
    44                     Thread.sleep(1000);
    45                 } catch (InterruptedException e) {
    46                     e.printStackTrace();
    47                 }
    48                 System.out.println(count);
    49             } catch (RuntimeException e) {
    50                 e.printStackTrace();
    51             }
    52             finally
    53             {
    54                 //lock.unlock();
    55                 sp.release();  //释放锁
    56             }
    57         }
    58     }        
    59     
    60 }

![复制代码][copycode.gif]

转载连接一：[http://www.iteye.com/topic/980944][http_www.iteye.com_topic_980944]

CyclicBarrier介绍 （一）  
一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态，此屏障操作 很有用。  
主要方法：

1.  //设置parties、count及barrierCommand属性。 
2.  CyclicBarrier(**int**):
3.  
4.  //当await的数量到达了设定的数量后，首先执行该Runnable对象。 
5.  CyclicBarrier(**int**,Runnable):
6.  
7.  //通知barrier已完成线程 
8.  await():

//设置parties、count及barrierCommand属性。
    CyclicBarrier(int):
    
    //当await的数量到达了设定的数量后，首先执行该Runnable对象。
    CyclicBarrier(int,Runnable):
    
    //通知barrier已完成线程
    await():

应用场景  
在某种需求中，比如一个大型的任务，常常需要分配好多子任务去执行，只有当所有子任务都执行完成时候，才能执行主任务，这时候，就可以选择CyclicBarrier了。  
  
实例分析  
我们需要统计全国的业务数据。其中各省的数据库是独立的，也就是说按省分库。并且统计的数据量很大，统计过程也比较慢。为了提高性能，快速计算。我们采取并发的方式，多个线程同时计算各省数据，最后再汇总统计。在这里CyclicBarrier就非常有用。看代码：  
  
主要类：

1.  /\*\* 
2.   \* 各省数据独立，分库存偖。为了提高计算性能，统计时采用每个省开一个线程先计算单省结果，最后汇总。 
3.   \* 
4.   \* @author guangbo email:weigbo@163.com 
5.   \* 
6.   \*/
7.  **public** **class** Total \{
8.  
9.  // private ConcurrentHashMap result = new ConcurrentHashMap(); 
10. 
11. **public** **static** **void** main(String\[\] args) \{
12. TotalService totalService = **new** TotalServiceImpl();
13. CyclicBarrier barrier = **new** CyclicBarrier(5,
14. **new** TotalTask(totalService));
15. 
16. // 实际系统是查出所有省编码code的列表，然后循环，每个code生成一个线程。 
17. **new** BillTask(**new** BillServiceImpl(), barrier, "北京").start();
18. **new** BillTask(**new** BillServiceImpl(), barrier, "上海").start();
19. **new** BillTask(**new** BillServiceImpl(), barrier, "广西").start();
20. **new** BillTask(**new** BillServiceImpl(), barrier, "四川").start();
21. **new** BillTask(**new** BillServiceImpl(), barrier, "黑龙江").start();
22. 
23. \}
24. \}
25. 
26. /\*\* 
27.  \* 主任务：汇总任务 
28.  \*/
29. **class** TotalTask **implements** Runnable \{
30. **private** TotalService totalService;
31. 
32. TotalTask(TotalService totalService) \{
33. **this**.totalService = totalService;
34. \}
35. 
36. **public** **void** run() \{
37. // 读取内存中各省的数据汇总，过程略。 
38. totalService.count();
39. System.out.println("=======================================");
40. System.out.println("开始全国汇总");
41. \}
42. \}
43. 
44. /\*\* 
45.  \* 子任务：计费任务 
46.  \*/
47. **class** BillTask **extends** Thread \{
48. // 计费服务 
49. **private** BillService billService;
50. **private** CyclicBarrier barrier;
51. // 代码，按省代码分类，各省数据库独立。 
52. **private** String code;
53. 
54. BillTask(BillService billService, CyclicBarrier barrier, String code) \{
55. **this**.billService = billService;
56. **this**.barrier = barrier;
57. **this**.code = code;
58. \}
59. 
60. **public** **void** run() \{
61. System.out.println("开始计算--" \+ code + "省--数据！");
62. billService.bill(code);
63. // 把bill方法结果存入内存，如ConcurrentHashMap,vector等,代码略 
64. System.out.println(code + "省已经计算完成,并通知汇总Service！");
65. **try** \{
66. // 通知barrier已经完成 
67. barrier.await();
68. \} **catch** (InterruptedException e) \{
69. e.printStackTrace();
70. \} **catch** (BrokenBarrierException e) \{
71. e.printStackTrace();
72. \}
73. \}
74. 
75. \}

结果：  
开始计算--北京省--数据！  
开始计算--上海省--数据！  
北京省已经计算完成,并通知汇总Service！  
开始计算--四川省--数据！  
四川省已经计算完成,并通知汇总Service！  
上海省已经计算完成,并通知汇总Service！  
开始计算--广西省--数据！  
广西省已经计算完成,并通知汇总Service！  
开始计算--黑龙江省--数据！  
黑龙江省已经计算完成,并通知汇总Service！  
=======================================  
开始全国汇总  
CyclicBarrier介绍 （二）

张孝祥视频学习笔记：

CyclicBarrier 表示大家彼此等待，大家集合好后才开始出发，分散活动后又在i指定地点集合碰面，这就好比整个公司的人员利用周末时间集体郊游一样，先各自从家出发到公司集合后，再同时出发到公园游玩，在指定地点集合后再同时开始就餐……

import java.util.concurrent.BrokenBarrierException;  
import java.util.concurrent.CyclicBarrier;  
import java.util.concurrent.ExecutorService;  
import java.util.concurrent.Executors;

public class CyclicBarrierTest \{  
public static void main(String \[\] args)\{  
ExecutorService service=Executors.newCachedThreadPool();  
final CyclicBarrier cb=new CyclicBarrier(3); //三个线程同时到达  
for(int i=0;i<3;i++)\{  
Runnable runnable=new Runnable()\{  
public void run()\{  
try \{  
Thread.sleep((long)(Math.random()\*10000));  
System.out.println("线程"+Thread.currentThread().getName()+  
"即将到达集合地点1，当前已有"+(cb.getNumberWaiting()+1)+"个已到达"+  
(cb.getNumberWaiting()==2?"都到齐了，继续走啊":"正在等候"));  
try \{  
cb.await();  
\} catch (BrokenBarrierException e) \{  
// TODO Auto-generated catch block  
e.printStackTrace();  
\}  
Thread.sleep((long)(Math.random()\*10000));  
System.out.println("线程"+Thread.currentThread().getName()+  
"即将到达集合地点2，当前已有"+(cb.getNumberWaiting()+1)+"个已到达"+  
(cb.getNumberWaiting()==2?"都到齐了，继续走啊":"正在等候"));  
try \{  
cb.await();  
\} catch (BrokenBarrierException e) \{  
// TODO Auto-generated catch block  
e.printStackTrace();  
\}  
Thread.sleep((long)(Math.random()\*10000));  
System.out.println("线程"+Thread.currentThread().getName()+  
"即将到达集合地点3，当前已有"+(cb.getNumberWaiting()+1)+"个已到达"+  
(cb.getNumberWaiting()==2?"都到齐了，继续走啊":"正在等候"));  
try \{  
cb.await();  
\} catch (BrokenBarrierException e) \{  
// TODO Auto-generated catch block  
e.printStackTrace();  
\}  
\} catch (InterruptedException e) \{  
// TODO Auto-generated catch block  
e.printStackTrace();  
\}  
\}  
\};  
service.execute(runnable);  
\}  
service.shutdown();  
\}  
\}

运行结果：

线程pool-1-thread-3即将到达集合地点1，当前已有1个已到达正在等候  
线程pool-1-thread-2即将到达集合地点1，当前已有2个已到达正在等候  
线程pool-1-thread-1即将到达集合地点1，当前已有3个已到达都到齐了，继续走啊  
线程pool-1-thread-1即将到达集合地点2，当前已有1个已到达正在等候  
线程pool-1-thread-2即将到达集合地点2，当前已有2个已到达正在等候  
线程pool-1-thread-3即将到达集合地点2，当前已有3个已到达都到齐了，继续走啊  
线程pool-1-thread-2即将到达集合地点3，当前已有1个已到达正在等候  
线程pool-1-thread-1即将到达集合地点3，当前已有2个已到达正在等候  
线程pool-1-thread-3即将到达集合地点3，当前已有3个已到达都到齐了，继续走啊

# [java.util.concurrent.Exchanger应用范例与原理浅析--转载][java.util.concurrent.Exchanger_--] #

一、简介  
Exchanger是自jdk1.5起开始提供的工具套件，一般用于两个工作线程之间交换数据。在本文中我将采取由浅入深的方式来介绍分析这个工具类。首先我们来看看官方的api文档中的叙述：

A synchronization point at which threads can pair and swap elements within pairs. Each thread presents some object on entry to the exchange method, matches with a partner thread, and receives its partner's object on return. An Exchanger may be viewed as a bidirectional form of a SynchronousQueue. Exchangers may be useful in applications such as genetic algorithms and pipeline designs.

在以上的描述中，有几个要点：

*  此类提供对外的操作是同步的；
 *  用于成对出现的线程之间交换数据；
 *  可以视作双向的同步队列；
 *  可应用于基因算法、流水线设计等场景。

接着看api文档，这个类提供对外的接口非常简洁，一个无参构造函数，两个重载的范型exchange方法：  
public V exchange(V x) throws InterruptedException  
public V exchange(V x, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException  
从官方的javadoc可以知道，当一个线程到达exchange调用点时，如果它的伙伴线程此前已经调用了此方法，那么它的伙伴会被调度唤醒并与之进行对象交换，然后各自返回。如果它的伙伴还没到达交换点，那么当前线程将会被挂起，直至伙伴线程到达——完成交换正常返回；或者当前线程被中断——抛出中断异常；又或者是等候超时——抛出超时异常。  
  
二、一个简单的例子  
按照某大师的观点，行为知之先，在知道了Exchanger的大致用途并参阅了使用说明后，我们马上动手写个例子来跑一跑：

![复制代码][copycode.gif]

import java.util.concurrent.Exchanger;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    import org.apache.log4j.Logger;
    
    /**
     * @Title: ExchangerTest
     * @Description: Test class for Exchanger
     * @Company: CSAIR
     * @Author: lixuanbin
     * @Creation: 2014年12月14日
     * @Version:1.0
     */
    public class ExchangerTest {
        protected static final Logger log = Logger.getLogger(ExchangerTest.class);
        private static volatile boolean isDone = false;
    
        static class ExchangerProducer implements Runnable {
            private Exchanger<Integer> exchanger;
            private static int data = 1;
            ExchangerProducer(Exchanger<Integer> exchanger) {
                this.exchanger = exchanger;
            }
    
            @Override
            public void run() {
                while (!Thread.interrupted() && !isDone) {
                    for (int i = 1; i <= 3; i++) {
                        try {
                            TimeUnit.SECONDS.sleep(1);
                            data = i;
                            System.out.println("producer before: " + data);
                            data = exchanger.exchange(data);
                            System.out.println("producer after: " + data);
                        } catch (InterruptedException e) {
                            log.error(e, e);
                        }
                    }
                    isDone = true;
                }
            }
        }
    
        static class ExchangerConsumer implements Runnable {
            private Exchanger<Integer> exchanger;
            private static int data = 0;
            ExchangerConsumer(Exchanger<Integer> exchanger) {
                this.exchanger = exchanger;
            }
    
            @Override
            public void run() {
                while (!Thread.interrupted() && !isDone) {
                    data = 0;
                    System.out.println("consumer before : " + data);
                    try {
                        TimeUnit.SECONDS.sleep(1);
                        data = exchanger.exchange(data);
                    } catch (InterruptedException e) {
                        log.error(e, e);
                    }
                    System.out.println("consumer after : " + data);
                }
            }
        }
    
        /**
         * @param args
         */
        public static void main(String[] args) {
            ExecutorService exec = Executors.newCachedThreadPool();
            Exchanger<Integer> exchanger = new Exchanger<Integer>();
            ExchangerProducer producer = new ExchangerProducer(exchanger);
            ExchangerConsumer consumer = new ExchangerConsumer(exchanger);
            exec.execute(producer);
            exec.execute(consumer);
            exec.shutdown();
            try {
                exec.awaitTermination(30, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                log.error(e, e);
            }
        }
    }

![复制代码][copycode.gif]

这大致可以看作是一个简易的生产者消费者模型，有两个任务类，一个递增地产生整数，一个产生整数0，然后双方进行交易。每次交易前的生产者和每次交易后的消费者都会sleep 1秒来模拟数据处理的消耗，并在交易前后把整数值打印到控制台以便检测结果。在这个例子里交易循环只执行三次，采用一个volatile boolean来控制交易双方线程的退出。  
我们来看看程序的输出：

consumer before : 0  
producer before: 1  
consumer after : 1  
producer after: 0  
consumer before : 0  
producer before: 2  
producer after: 0  
consumer after : 2  
consumer before : 0  
producer before: 3  
producer after: 0  
consumer after : 3

输出结果验证了以下两件事情：

*  exchange方法真的帮一对线程交换了数据；
 *  exchange方法真的会阻塞调用方线程直至另一方线程参与交易。

那么在中断和超时两种情况下程序的运行表现会是怎样呢？作为一个小练习，有兴趣的观众可以设想并编写测试用例覆盖验证之。接下来谈谈最近我在生产场景中对Exchanger的应用。

三、实战场景  
1.问题描述  
最近接到外部项目组向我组提出的接口需求，需要查询我们业务办理量的统计情况。我们系统目前的情况是，有一个日增长十多万、总数据量为千万级别的业务办理明细表（xxx\_info），每人次的业务办理结果会实时写入其中。以往对外提供的业务统计接口是在每次被调用时候在明细表中执行SQL查询（select、count、where、group by等），响应时间很长，对原生产业务的使用也有很大的影响。于是我决定趁着这次新增接口的上线机会对系统进行优化。  
2.优化思路  
首先是在明细表之外再建立一个数据统计（xxx\_statistics）表，考虑到目前数据库的压力以及公司内部质管流控等因素，暂没有分库存放，仍旧与原明细表放在同一个库。再设置一个定时任务于每日凌晨对明细表进行查询、过滤、统计、排序等操作，把统计结果插入到统计表中。然后对外暴露统计接口查询统计报表。现在的设计与原来的实现相比，虽然牺牲了统计表所占用的少量额外的存储空间（每日新增的十来万条业务办理明细记录经过处理最终会变成几百条统计表的记录），但是却能把select、count这样耗时的数据统计操作放到凌晨时段执行以避开白天的业务办理高峰，分表处理能够大幅降低对生产业务明细表的性能影响，而对外提供的统计接口的查询速度也将得到几个数量级的提升。当然，还有一个缺点是，不能实时提供当天的统计数据，不过这也是双方可以接受的。  
3.设计实现  
设计一个定时任务，每日凌晨执行。在定时任务中启动两个线程，一个线程负责对业务明细表（xxx\_info）进行查询统计，把统计的结果放置在内存缓冲区，另一个线程负责读取缓冲区中的统计结果并插入到业务统计表（xxx\_statistics）中。  
亲，这样的场景是不是听起来很有感觉？没错!两个线程在内存中批量交换数据，这个事情我们可以使用Exchanger去做！我们马上来看看代码如何实现。

生产者线程：

![复制代码][copycode.gif]

class ExchangerProducer implements Runnable {
        private Exchanger<Set<XXXStatistics>> exchanger;
        private Set<XXXStatistics> holder;
        private Date fltDate;
        private int threshold;
    
        ExchangerProducer(Exchanger<Set<XXXStatistics>> exchanger,
                Set<XXXStatistics> holder, Date fltDate, int threshold) {
            this.exchanger = exchanger;
            this.holder = holder;
            this.fltDate = fltDate;
            this.threshold = threshold;
        }
    
        @Override
        public void run() {
            try {
                while (!Thread.interrupted() && !isDone) {
                    List<XXXStatistics> temp1 = null;
                    List<XXXStatistics> temp11 = null;
                    for (int i = 0; i < allCities.size(); i++) {
                        try {
                            temp1 = xxxDao
                                    .findStatistics1(
                                            fltDate, allCities.get(i));
                            temp11 = xxxDao
                                    .findStatistics2(
                                            fltDate, allCities.get(i),
                                            internationalList);
                            if (temp1 != null && !temp1.isEmpty()) {
                                calculationCounter.addAndGet(temp1.size());
                                if (temp11 != null && !temp11.isEmpty()) {
                                    // merge two lists into temp1
                                    mergeLists(temp1, temp11);
                                    temp11.clear();
                                    temp11 = null;
                                }
                                // merge temp1 into holder set
                                mergeListToSet(holder, temp1);
                                temp1.clear();
                                temp1 = null;
                            }
                        } catch (Exception e) {
                            log.error(e, e);
                        }
                        // Insert every ${threshold} or the last into database.
                        if (holder.size() >= threshold
                                || i == (allCities.size() - 1)) {
                            log.info("data collected: \n" + holder);
                            holder = exchanger.exchange(holder);
                            log.info("data submitted");
                        }
                    }
                    // all cities are calculated
                    isDone = true;
                }
                log.info("calculation job done, calculated: "
                        + calculationCounter.get());
            } catch (InterruptedException e) {
                log.error(e, e);
            }
            exchanger = null;
            holder.clear();
            holder = null;
            fltDate = null;
        }
    }

![复制代码][copycode.gif]

代码说明：

*  threshold：缓冲区的容量阀值；
 *  allCities：城市列表，迭代这个列表作为入参来执行查询统计；
 *  XXXStatistics：统计数据封装实体类，实现了Serializable和Comparable接口，覆写equals和compareTo方法，以利用TreeSet提供的去重和排序处理；
 *  isDone：volatile boolean，标识统计任务是否完成；
 *  holder：TreeSet<XXXStatistics>，存放统计结果的内存缓冲区，容量达到阀值后提交给Exchanger执行exchange操作；
 *  dao.findStatistics1，dao.findStatistics2：简化的数据库查询统计操作，此处仅供示意；
 *  calculationCounter：AtomicInteger，标记生产端所提交的记录总数；
 *  mergeLists，mergeListToSet：内部私有工具方法，把dao查询返回的列表合并到holder中；

消费者线程：

![复制代码][copycode.gif]

class ExchangerConsumer implements Runnable {
        private Exchanger<Set<XXXStatistics>> exchanger;
        private Set<XXXStatistics> holder;
    
        ExchangerConsumer(Exchanger<Set<XXXStatistics>> exchanger,
                Set<XXXStatistics> holder) {
            this.exchanger = exchanger;
            this.holder = holder;
        }
    
        @Override
        public void run() {
            try {
                List<XXXStatistics> tempList;
                while (!Thread.interrupted() && !isDone) {
                    holder = exchanger.exchange(holder);
                    log.info("got data: \n" + holder);
                    if (holder != null && !holder.isEmpty()) {
                        try {
                            // insert data into database
                            tempList = convertSetToList(holder);
                            insertionCounter.addAndGet(xxxDao
                                    .batchInsertXXXStatistics(tempList));
                            tempList.clear();
                            tempList = null;
                        } catch (Exception e) {
                            log.error(e, e);
                        }
                        // clear the set
                        holder.clear();
                    } else {
                        log.info("wtf, got an empty list");
                    }
                    log.info("data processed");
                }
                log.info("insert job done, inserted: " + insertionCounter.get());
            } catch (InterruptedException e) {
                log.error(e, e);
            }
            exchanger = null;
            holder.clear();
            holder = null;
        }
    }

![复制代码][copycode.gif]

代码说明：

*  convertSetToList：由于dao接口的限制，需把交换得到的Set转换为List；
 *  batchInsertXXXStatistics：使用jdbc4的batch update而实现的批量插入dao接口；
 *  insertionCounter：AtomicInteger，标记消费端插入成功的记录总数；

调度器代码：

代码说明：  
调度器的代码就四个步骤：初始化、提交任务并等候处理结果、清理、返回。初始化阶段使用了jdk提供的线程池提交生产者和消费者任务，设置了最长等候时间calculationTimeoutMinutes，如果调度器线程被中断或者任务执行超时，awaitTermination会返回false，此时就强行关闭线程池并记录到日志。统计操作每日凌晨执行一次，所以在任务退出前的清理阶段建议jvm执行gc以尽早释放计算时所产生的垃圾对象。在结果返回阶段，如果查询统计出来的记录条数和插入成功的条数相等则返回true，否则返回false。

4.小结  
在这个案例中，使用Exchanger进行批量的双向数据交换可谓恰如其分：生产者在执行新的查询统计任务填入数据到缓冲区的同时，消费者正在批量插入生产者换入的上一次产生的数据，系统的吞吐量得到平滑的提升；计算复杂度、内存消耗、系统性能也能通过相关的参数设置而得到有效的控制（在消费端也可以对holder进行再次分割以控制每次批插入的大小，建议参阅数据库厂商以及数据库驱动包的说明文档以确定jdbc的最优batch update size）；代码的实现也很简洁易懂。这些优点，是采用有界阻塞队列所难以达到的。  
程序的输出结果与业务紧密相关，就不打印出来了。可以肯定的是，经过了一段时间的摸索调优，内存消耗、执行速度和处理结果还是比较满意的。

原文地址：http://lixuanbin.iteye.com/blog/2166772

[copycode.gif]: /images/20220731/d415f66ab29542a4817251ff7f6a54f6.png
[http_www.iteye.com_topic_980944]: http://www.iteye.com/topic/980944
[java.util.concurrent.Exchanger_--]: http://www.cnblogs.com/davidwang456/p/4179488.html