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

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

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

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

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

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

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

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

?? 生产者线程：

<div class="cnblogs_code">

 ExchangerProducer  Exchanger> Set ExchangerProducer(Exchanger</span><Set<XXXStatistics>><span style="color: #000000;"&gt; exchanger,Set</span><XXXStatistics> holder,Date fltDate,<span style="color: #0000ff;"&gt;int</span><span style="color: #000000;"&gt; threshold) {
    </span><span style="color: #0000ff;"&gt;this</span>.exchanger =<span style="color: #000000;"&gt; exchanger;
    </span><span style="color: #0000ff;"&gt;this</span>.holder =<span style="color: #000000;"&gt; holder;
    </span><span style="color: #0000ff;"&gt;this</span>.fltDate =<span style="color: #000000;"&gt; fltDate;
    </span><span style="color: #0000ff;"&gt;this</span>.threshold =<span style="color: #000000;"&gt; threshold;
}

@Override
</span><span style="color: #0000ff;"&gt;public</span> <span style="color: #0000ff;"&gt;void</span><span style="color: #000000;"&gt; run() {
    </span><span style="color: #0000ff;"&gt;try</span><span style="color: #000000;"&gt; {
        </span><span style="color: #0000ff;"&gt;while</span> (!Thread.interrupted() &amp;&amp; !<span style="color: #000000;"&gt;isDone) {
            List</span><XXXStatistics> temp1 = <span style="color: #0000ff;"&gt;null</span><span style="color: #000000;"&gt;;
            List</span><XXXStatistics> temp11 = <span style="color: #0000ff;"&gt;null</span><span style="color: #000000;"&gt;;
            </span><span style="color: #0000ff;"&gt;for</span> (<span style="color: #0000ff;"&gt;int</span> i = 0; i < allCities.size(); i++<span style="color: #000000;"&gt;) {
                </span><span style="color: #0000ff;"&gt;try</span><span style="color: #000000;"&gt; {
                    temp1 </span>=<span style="color: #000000;"&gt; xxxDao
                            .findStatistics1(
                                    fltDate,allCities.get(i));
                    temp11 </span>=<span style="color: #000000;"&gt; xxxDao
                            .findStatistics2(
                                    fltDate,allCities.get(i),internationalList);
                    </span><span style="color: #0000ff;"&gt;if</span> (temp1 != <span style="color: #0000ff;"&gt;null</span> &amp;&amp; !<span style="color: #000000;"&gt;temp1.isEmpty()) {
                        calculationCounter.addAndGet(temp1.size());
                        </span><span style="color: #0000ff;"&gt;if</span> (temp11 != <span style="color: #0000ff;"&gt;null</span> &amp;&amp; !<span style="color: #000000;"&gt;temp11.isEmpty()) {
                            </span><span style="color: #008000;"&gt;//</span><span style="color: #008000;"&gt; merge two lists into temp1</span>

<span style="color: #000000;"> mergeLists(temp1,temp11);
temp11.clear();
temp11 = <span style="color: #0000ff;">null<span style="color: #000000;">;
}
<span style="color: #008000;">//<span style="color: #008000;"> merge temp1 into holder set
<span style="color: #000000;"> mergeListToSet(holder,temp1);
temp1.clear();
temp1 = <span style="color: #0000ff;">null<span style="color: #000000;">;
}
} <span style="color: #0000ff;">catch<span style="color: #000000;"> (Exception e) {
log.error(e,e);
}
<span style="color: #008000;">//<span style="color: #008000;"> Insert every ${threshold} or the last into database.
<span style="color: #0000ff;">if (holder.size() >=<span style="color: #000000;"> threshold
|| i == (allCities.size() - 1<span style="color: #000000;">)) {
log.info("data collected: n" +<span style="color: #000000;"> holder);
holder =<span style="color: #000000;"> exchanger.exchange(holder);
log.info("data submitted"<span style="color: #000000;">);
}
}
<span style="color: #008000;">//<span style="color: #008000;"> all cities are calculated
isDone = <span style="color: #0000ff;">true<span style="color: #000000;">;
}
log.info("calculation job done,calculated: "
+<span style="color: #000000;"> calculationCounter.get());
} <span style="color: #0000ff;">catch<span style="color: #000000;"> (InterruptedException e) {
log.error(e,e);
}
exchanger = <span style="color: #0000ff;">null<span style="color: #000000;">;
holder.clear();
holder = <span style="color: #0000ff;">null<span style="color: #000000;">;
fltDate = <span style="color: #0000ff;">null<span style="color: #000000;">;
}
}