2019独角兽企业重金招聘Python工程师标准>>>

一个countDown在多线程调度下使用不当的分享

1. 诡异的数据抖动

在一个需求开发过程中，由于有多角色需要获取每个角色下的菜单；结果出现了单角色下拉去菜单没问题，多角色情况下只有一个角色的菜单正常返回的问题。这个问题很忧伤，没有菜单如何进功能页面？

2. 怀疑是缓存

因为多角色下菜单采用了Redis缓存，故而怀疑是其中一个角色下的菜单是缓存失效，但是关闭掉缓存依然不起作用，排除缓存影响。

3. debug发现问题

通过增加日志输出，在关闭掉缓存的实时模式下，依然存在菜单时而有，时而没有的情况。证明应该是代码有问题。

在一个多线程调度调度服务类中，发现一个问题，即在debug到如下代码，会出现后续代码未执行完全，接口结果即被返回的情况。

//经过查询相关API，不会出现时序问题，可放心使用final CountDownLatch latch = new CountDownLatch(callableList.size());for(Callable callable :callableList){ListenableFuture<T> listenableFuture = threadPoolTaskExecutor.submitListenable(callable);listenableFuture.addCallback(new ListenableFutureCallback<T>() {@Overridepublic void onFailure(Throwable throwable) {//过早调用countDown BUGlatch.countDown();LogHelper.EXCEPTION.error("执行任务异常",throwable);if(futureCallback!=null){futureCallback.onFailure(throwable);}}@Overridepublic void onSuccess(T t) {//过早调用countDown BUGlatch.countDown();if(futureCallback!=null){futureCallback.onSuccess(t);}}});}

4. countDown调用时机不对

在调用远程接口返回后，立即执行 lacth.countDown(); 会立刻造成主线程阻塞释放，立即响应结果，丢失部分数据。如下图所示，在第二个任务获取数据处理完成后，就立即调用latch.countDown(), 致使后续的回调还未执行。主线程在收到 latch的释放阻塞后，返回了不完整的数据结果。

错误的调用时序

改造后，将countDown放在回调执行完成之后，并放置在 try {} finally { }代码块之中，保证一定得到执行，防止抛异常后，主线程阻塞。如下所示：

        final CountDownLatch latch = new CountDownLatch(callableList.size());for(Callable callable :callableList){ListenableFuture<T> listenableFuture = threadPoolTaskExecutor.submitListenable(callable);listenableFuture.addCallback(new ListenableFutureCallback<T>() {@Overridepublic void onFailure(Throwable throwable) {try{LogHelper.DEFAULT.info("多线程回调,latchCount="+latch.getCount());LogHelper.EXCEPTION.error("执行任务异常",throwable);if(futureCallback!=null){futureCallback.onFailure(throwable);}}finally {latch.countDown();}}@Overridepublic void onSuccess(T t) {try{LogHelper.DEFAULT.info("多线程回调成功，latchCount="+latch.getCount());if(futureCallback!=null){futureCallback.onSuccess(t);}}finally {latch.countDown();}}});}

5. 本次使用的多线程调度说明

使用Future阻塞模式，不会出现以上问题，使用future.get()的阻塞式获取，不需要CountDownLatch工具类配合使用。而且本次其实也可以使用Future来实现同样的功能。但是Future没有提供 onFailure, onSucess 这样的回调接口，考虑到易用性，采用了ListenableFuture；

本次采用的 spring的 ListenableFuture 方式回调来实现回调式聚合。在对CountDownLatch使用不当的情况下，出现了该问题。解决该问题后，功能运行正常。

本次提供的服务类 ThreadPoolExecutorService, 其主要的方法如下：


/*** 线程池服务类** @author David* @since 2018/5/15*/
public interface ThreadPoolExecutorService {/*** 执行多线程任务处理，并通过 futureCallback对结果进行回调处理* @param callableList  异步线程可执行的任务 List* @param futureCallback  异步回调* @param timeout 超时时间，毫秒* @param <T>*/<T> void execute(List<Callable<T>> callableList, ListenableFutureCallback<T> futureCallback, long timeout);/*** 执行多线程任务处理，并将结果聚合成一个List 进行返回** @param callableList   异步线程可执行的任务 List* @param failureCallback  失败的回调方法* @param skipNull        是否忽略Null 结果，如果忽略 null 不会添加到List 之中* @param timeout 超时时间，毫秒* @param <T>*/<T> List<T> executeAndMerge(List<Callable<T>> callableList, FailureCallback failureCallback, boolean skipNull, long timeout);/*** 执行多线程任务处理，并将结果List,聚合成一个List 进行返回** @param callableList 异步线程可执行的任务 List* @param failureCallback 失败的回调方法* @param timeout 超时时间，毫秒* @param <T>*/<T> List<T> executeAndMergeList(List<Callable<List<T>>> callableList, FailureCallback failureCallback, long timeout);

6. Future 和 ListenableFuture的区别

spring 或者 guava 的 ListenableFutrue其使用方式和机理应该是类似的，这里的说明是通用的。本次工具类使用的是spring自带的ListenableFutrue。

6.1 区别说明

ListenableFuture顾名思义就是可以监听的Future，它是对java原生Future的扩展增强。我们知道Future表示一个异步计算任务，当任务完成时可以得到计算结果。如果我们希望一旦计算完成就拿到结果展示给用户或者做另外的计算，就必须使用另一个线程不断的查询计算状态。这样做，代码复杂，而且效率低下。使用ListenableFuture帮我们检测Future是否完成了，如果完成就自动调用回调函数，这样可以让主线程不必阻塞，减少并发程序的复杂度。

6.4 spring ListenableFuture使用示例

一个简单的示例，如果要获得 ListenableFuture，则需要一个对Java线程池进行修饰过的线程池执行器。如下所示：

<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"><property name="corePoolSize" value="${threadpool.corePoolSize}" /><property name="keepAliveSeconds" value="${threadpool.keepAliveSeconds}" /><property name="maxPoolSize" value="${threadpool.maxPoolSize}" /><property name="queueCapacity" value="${threadpool.queueCapacity}" /><property name="rejectedExecutionHandler"><bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" /></property></bean>

guava的请参考以下方式进行修饰，这里不进行详细说明：

   //Java线程池的类型是可选的【根据场景自行构建】   ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());

得到修饰过的线程池执行器后，即可提交可Callable任务，得到 ListenableFuture 进行处理。

以下为计算1~5的3次方的结果并输出，不需要聚合结果，每个线程计算完毕后，立刻输出结果：

for(int i=1; i<=5; i++){final int num = i;ListenableFuture<Integer> listenableFuture = threadPoolTaskExecutor.submitListenable(new Callable<Integer>() {@Overridepublic Integer call(){return (int)Math.pow(num,3);}});listenableFuture.addCallback(new ListenableFutureCallback<Integer>() {@Overridepublic void onFailure(Throwable throwable) {LogHelper.EXCEPTION.error("处理失败", throwable);}@Overridepublic void onSuccess(Integer result) {System.out.println(num + "的3次方为："+ result);}});}

6.2 是否可以添加多个callback

答案是肯定的，因为ListenableFuture 的addCallback是添加到ListenableFutureCallbackRegistry 一个注册中心。而注册中心底层是支持多个回调的。在6.3会具体介绍回调方法注册中心的处理逻辑。

 public void addCallback(ListenableFutureCallback<? super T> callback) {this.callbacks.addCallback(callback);}

6.3 addCallback是否需要考虑时序

guava 和 spring的 ListenableFuture 均做了时序兼容，在listenableFuture执行的任意时刻调用 addCallback 均可准确的执行回调。这里就不得不说在调用addCallback的时候，其实将回到方法注册到ListenableFutureCallbackRegistry（回调注册中心）。