19 | CountDownLatch和CyclicBarrier：如何让多线程步调一致？

王宝令

前几天老板突然匆匆忙忙过来，说对账系统最近越来越慢了，能不能快速优化一下。我了解了对账系统的业务后，发现还是挺简单的，用户通过在线商城下单，会生成电子订单，保存在订单库；之后物流会生成派送单给用户发货，派送单保存在派送单库。为了防止漏派送或者重复派送，对账系统每天还会校验是否存在异常订单。
对账系统的处理逻辑很简单，你可以参考下面的对账系统流程图。目前对账系统的处理逻辑是首先查询订单，然后查询派送单，之后对比订单和派送单，将差异写入差异库。
对账系统流程图
对账系统的代码抽象之后，也很简单，核心代码如下，就是在一个单线程里面循环查询订单、派送单，然后执行对账，最后将写入差异库。
while(存在未对账订单){
  // 查询未对账订单
  pos = getPOrders();
  // 查询派送单
  dos = getDOrders();
  // 执行对账操作
  diff = check(pos, dos);
  // 差异写入差异库
  save(diff);
} 
利用并行优化对账系统老板要我优化性能，那我就首先要找到这个对账系统的瓶颈所在。
目前的对账系统，由于订单量和派送单量巨大，所以查询未对账订单 getPOrders() 和查询派送单 getDOrders() 相对较慢，那有没有办法快速优化一下呢？目前对账系统是单线程执行的，图形化后是下图这个样子。对于串行化的系统，优化性能首先想到的是能否利用多线程并行处理。
对账系统单线程执行示意图
所以，这里你应该能够看出来这个对账系统里的瓶颈：查询未对账订单 getPOrders() 和查询派送单 getDOrders() 是否可以并行处理呢？显然是可以的，因为这两个操作并没有先后顺序的依赖。这两个最耗时的操作并行之后，执行过程如下图所示。对比一下单线程的执行示意图，你会发现同等时间里，并行执行的吞吐量近乎单线程的 2 倍，优化效果还是相对明显的。

公开

同步至部落

取消

完成

0/2000

荧光笔

直线

曲线

笔记

复制

AI

深入了解
翻译
英语
中文简体
中文繁体
法语
德语
日语
韩语
俄语
西班牙语
阿拉伯语
解释
总结

仅可试看部分内容，如需阅读全部内容，请付费购买文章所属专栏
《Java 并发编程实战》，新⼈⾸单¥59

立即购买

登录后留言

全部留言(155)

最新
精选

张申傲
我觉得老师的问题其实是两个: 1.为啥要用线程池，而不是在回调函数中直接调用？ 2.线程池为啥使用单线程的？我的考虑: 1.使用线程池是为了异步操作，否则回掉函数是同步调用的，也就是本次对账操作执行完才能进行下一轮的检查。 2.线程数量固定为1，防止了多线程并发导致的数据不一致，因为订单和派送单是两个队列，只有单线程去两个队列中取消息才不会出现消息不匹配的问题。
作者回复: 👍
13
249
J.M.Liu
老师，CyclicBarrier的回调函数在哪个线程执行啊？主线程吗？比如这里的最后一段代码中，循环会在回调的时候阻塞吗？如果是这样的话，那check函数岂不是可以直接作为回调函数了呀，并不需要线程池了啊
作者回复: 好问题，CyclicBarrier的回调函数执行在一个回合里最后执行await()的线程上，而且同步调用回调函数check()，调用完check之后，才会开始第二回合。所以check如果不另开一线程异步执行，就起不到性能优化的作用了。
8
180
undifined
线程池大小为1是必要的，如果设置为多个，有可能会两个线程 A 和 B 同时查询，A 的订单先返回，B 的派送单先返回，造成队列中的数据不匹配；所以1个线程实现生产数据串行执行，保证数据安全如果用Future 的话可以更方便一些： CompletableFuture<List> pOrderFuture = CompletableFuture.supplyAsync(this::getPOrders); CompletableFuture<List> dOrderFuture = CompletableFuture.supplyAsync(this::getDOrders); pOrderFuture.thenCombine(dOrderFuture, this::check) .thenAccept(this::save); 老师这样理解对吗，谢谢老师
作者回复: 对，👍👍👍
4
79
曾轼麟
老师推荐您使用ThreadPoolExecutor去实现线程池，并且实现里面的RejectedExecutionHandler和ThreadFactory，这样可以方便当调用订单查询和派送单查询的时候出现full gc的时候 dump文件可以快速定位出现问题的线程是哪个业务线程，如果是CountDownLatch，建议设置超时时间，避免由于业务死锁没有调用countDown()导致现线程睡死的情况
作者回复: 好建议，所有的阻塞操作，都需要设置超时时间，这是个很好的习惯。
4
65
空知
老师,关于CyclicBarrier回调函数,请教下自己写了个 CyclicBarrier的例子,回调函数总是在计数器归0时候执行,但是线程T1 T2要等回调函数执行结束之后才会再次执行...看了下CyclicBarrier 的源码,当内部计数器 index == 0时候, final Runnable command = barrierCommand; if (command != null) command.run(); 没有开启子线程吧.也就是说对账还是同步执行的,结束之后才是下一次的查询
作者回复: 所以才需要线程池来异步执行回调函数，你一不小心把答案找到了😂
7
65
波波
思考题中，如果生产者比较快，消费者比较慢，生产者通知的时候，消费者还在对账，这个时候会怎么处理？会不会导致消费者错失通知，导致队列满了，但是消费者却没有收到通知。
作者回复: 有这种可能，还能oom
7
27
nanquanmama
最后的那个例子，业务逻辑的部分已经变得很不直观，并发控制的逻辑掩盖住了业务逻辑。请问一下老师，实际项目开发中，并发控制逻辑如何做，才能和业务逻辑分离出来？
作者回复: 放到不同的类里，这方面传统的面向对象可以解决，lambda也能解决，这个模块的最后几章能解决你说的这个问题，但是更复杂的场景还得自己设计

19
... ...
追问：如果线程池是单线程的话。那假如生产者速度快运check函数执行时间。那是不是就会出现堵塞情况了。久而久之，是不是会出现队列内存溢出
作者回复: 会
3
17
Adam
如果生产者比较快，消费者check还没对账完会不会照成队列越来越多最后内存溢出了，有没有什么好的方案解决呢？
作者回复: 方案上基本都是限流

14
忍者无敌1995
有，如果为线程池有多个线程，则由于check()函数里面的两个remove并不是原子操作，可能导致消费错乱。假设订单队列中有P1，P2；派送队列中有D1,D2；两个线程T1,T2同时执行check，可能出现T1消费到P1,D2，T2消费到P2，D1，就是T1先执行pos.remove(0), 而后T2执行pos.remove(0);dos.remov(0);然后T1才执行dos.remove(0)的场景
作者回复: 多个线程有这个可能，所以线程池用的是单线程的
2
14

收起评论