如何让多线程步调一致？

前几天老板突然匆匆忙忙的过来说对账系统最近越来越慢了，能不能快速优化一下？我了解了对账系统的业务后，发现还是挺简单的，用户通过在线商城下单，会生成电子订单，保存在订单库。之后物流会生成派送单给用户发货，派送单保存在派送单库。为了防止漏送或重复派送，对账系统每天还会校验是否存在异常订单。
对账系统的处理逻辑很简单，你可以参考下面的对账系统流程图。目前的对账系统处理逻辑是，首先查询订单，然后查询派送单，然后对比订单和派送单，将差异写入差异库。
在这里插入图片描述
对账系统的代码抽象之后也很简单，核心代码如下，就是在一个单线程里循环查询订单、派送单，然后执行对账，最后写入差异库。

while(存在未对账的订单){//查询未对账订单pos = getPRders();//查询派送单dos = getDOrder();//执行对账操作diff = chech(pos,dos);//差异写入差异库save(diff);
}

利用并行优化对账系统

老板要我优化性能，那我就首先要找到这个对账系统的瓶颈所在。
目前的对账系统由于订单量和派送单量巨大，所以查询对账订单getPOrders和查询派送单getDOrders相对较慢，那有没有什么办法可以快速优化一下呢？目前对账系统是单线程执行的。图形化后是下面这个样子，对于串行化的系统性能优化，首先想到的是能否利用多线程并行处理。
在这里插入图片描述所以这里你应该所以，这里你应该能够看出来这个对账系统里的瓶颈，查询未对账订单getPOrder和查询派送订单getDOrder是否可以并行处理呢？显然是可以的，因为这两个操作并没有先后顺序的依赖，这两个最耗时的操作并行执行之后，执行过程如下图所示。对比一下单线程的执行示意图，你会发现同等时间里并行执行的吞吐量近乎单线程的两倍。提升效果还是相对明显的。
在这里插入图片描述
思路有了，下面我们再来看看如何利用代码实现。在下面的代码中，我们创建了两个线程T1和T2，并行执行查询未对账订单getPOrder和查询派送订单getDOrder这两个操作。在主线程中执行对账操作check和差异写入save两个操作，不过要注意的是，主线程需要等待线程T1和T2执行完才能执行check和save两个操作。为此，我们通过调用T1.join和T2.join来实现等待。当T1和T2线程退出时。调用T1.join和T2.join的主线程就会从阻塞状态被唤醒，从而执行之后的check和save。

while(存在未对账订单){//查询未对账订单Thread T1 = new Thread(() ->{pos = getPOrder();});T1.start();//查询派送订单Thread T2 = new Thread(()->{dos = getDOrder();});T2.start();//等待T1 T2的结果T1.join();T2.join();//执行对账操作diff = check(pos,dos);save(diff);
}

用CountDownLatch实现线程等待

经过上面的优化之后，基本可以和老板汇报收工了，但是有点美中不足，相信你也发现了，while循环里面每次都会创建新的线程，而创建线程可是个耗时的操作，所以最好是创建出来的线程能够循环利用。估计这时候你已经想到线程池了，是的，线程池就能解决这个问题。
而下面的代码就是利用线程池优化之后的，我们首先创建一个固定大小为2的线程池，之后在while循环里面重复利用，一切看上去都很顺利。但是有个问题好像无解了，那就是主线程如何知道getPOrder和getDOrder这两个操作什么时候执行完？前面的主线程通过调用线程T1和T2的join方法来等待线程T1和T2退出，但是在线程池的方案里，线程根本不会退出，所以join方法已经失效了。

//创建两个线程的线程池
Executor executor = Executor.newFixedThreadPool(2);
while(存在未对账订单){//查询未对账订单executor.execute(() ->{pos = getPOrder();});//查询派送订单executor.execute(()->{dos = getDOrder();});/* 如何实现线程等待呢？*///执行对账操作diff = check(pos,dos);save(diff);
}

那如何解决这个问题呢？你可以开动脑筋想出很多办法。最直接的办法是弄一个计数器，初始值设置成2，当执行完pos = getPOrder();这个操作之后，计数器减1，执行完dos = getDOrder();之后，计数器也减1，在主线程里，等待计数器等于零，当计数器等于零时，说明这两个查询操作执行完了。等待计数器等于零其实就是一个条件变量，用管程实现起来也很简单。
不过我并不建议你在实际项目中去实现上面的方案，因为Java并发包里已经提供了实现类似功能的工具类：CountDownLatch，这里我们可以直接使用。下面的代码示例中，在while循环里面，我们首先创建了一个CountDownLatch，计数器的初始值等于2。之后在pos = getPOrder();和dos = getDOrder();两个语句的后面对计数器执行减1操作。这个对计数器减1的操作是通过调用latch.countDown()来实现的。在主线程中，我们通过调用latch.await()来实现对计数器等于0的等待。

Executor executor = Executor.newFixedThreadPool(2);
while(存在未对账订单){//计数器初始化未2CountDownLatch latch = new CountDownLatch(2);//查询未对账订单executor.execute(() ->{pos = getPOrder();latch.countDown();});//查询派送订单executor.execute(()->{dos = getDOrder();latch.countDown()});//等待连个查询操作结束latch.await();//执行对账操作diff = check(pos,dos);save(diff);
}

进一步优化性能

经过上面的重重优化之后，长出一口气，终于可以交付了。不过在交付之前还需要再次审视一番，看还有没有优化的余地，仔细看还是有的。
前面我们将getPOrder和getDOrder这两个查询操作并行了，但这两个查询操作和对账操作check save之间还是串行的。很显然，这两个查询操作和对账操作也是可以并行的。也就是说，在执行对账操作的时候，可以同时去执行下一轮的查询操作，这个过程可以形象的描述为下面这幅图。
在这里插入图片描述
那接下来我们再来思考一下如何实现这步优化，两次查询操作都能够和对账操作并行。对账操作还依赖查询操作的结果，这明显有点生产者-消费者的意思。两次查询操作是生产者，对账操作是消费者。既然是生产者-消费者模型，那就需要有个队列来保存生产者生产数据，而消费者则从这个队列消费数据。
不过针对对账这个项目，我设计了两个队列，并且两个队列的元素之间还有对应关系，具体如下图所示，查询订单查询操作将订单查询结果插入订单队列，派送单查询操作将派送单插入派送单队列，这两个队列的元素之间是有一一对应的关系的。两个队列的好处是，对账操作可以每次从订单队列出一个元素，从派送单队列出一个元素，然后对这两个元素执行对账操作，这样数据一定不会乱掉。
在这里插入图片描述
下面再来看看如何利用双队列来实现完全的并行。一个最直接的想法是，一个线程T1执行订单的查询工作，一个线程T2执行派送单的查询工作，当线程T1和T2都各自生产完一条数据的时候，就通知线程T3执行对账操作。这个想法虽然看上去很简单，但其实还隐藏着一个条件，那就是线程T1和线程T2的工作步调要一致，不能一个跑的太快，一个跑的太慢。只有这样才能做到各自生产完一条数据的时候，通知线程T3。
下面这幅图形象的描述了上面的意图，线程T1和线程T2只有都生产完了一条数据的时候，才能一起向下执行，也就是说线程T1和线程T2还要相互等待，步调要一致。同时线程T1和T2都生产完一条数据的时候，还要能够通知线程T3执行对账操作。
在这里插入图片描述

用CyclicBarrier实现线程同步

下面我们就来实现上面提到的方案，这两个方案的难点有两个，一个是线程T1和T2要做到步调一致，另一个是要能够通知到线程T3。
你依然可以利用一个计数器来解决这两个难点。计数器初始化为2，线程T1和T2生产完一条数据，都将计数器减1，如果计数器大于0，则线程T1或T2等待，如果计数器等于零，则通知线程T3，并唤醒等待的线程T1和T2。与此同时，将计数器重新置为2。这样线程T1和T2生产下一条数据的时候，就可以继续使用这个计数器了。
同样，还是建议你不要在实际项目中这么做，因为Java并发包里也已经提供了相关的工具类：CyclicBarrier。下面的代码中，我们首先创建了一个计数器初始值为2的CyclicBarrier，你需要注意的是创建CyclicBarrier的时候我们还需要传入一个回调函数，当计数器减到0的时候会调用这个回调函数。
线程T1负责查询订单，当查询出一条的时候调用barrier.await()来将计数器减1，同时等待计数器变成0。线程T2负责查询派送订单，当查询出一条时也调用barrier.await()来将计数器减1，同时等待计数器变成0，当T1和T2都调用barrier.await()的时候，计数器会减到0，当T1和T2，此时T1和T2就可以执行下一条语句了，同时还会调用的回调函数来执行对账操作。
非常值得一提的是，CyclicBarrier的计数器有自动重置功能，当减到零的时候会自动重置回你设置的初始值，这个功能看用起来实在太方便了。

//订单队列
Vector<P> pos;
//派送单队列
Vector<P> dos;
//执行回调的线程池
Executor executor = Executor.newFixedThreadPool(1);
final CyclicBarrier barrier = new CyclicBarrier(2,()->{executor.execute(() -> check());});void check(){P p = pos.remove(0);D d = dos.remove(0);//执行对账操作diff = check(p,d);save(diff);
}void checkAll(){//查询未对账订单Thread T1 = new Thread(() ->{while(存在未对账订单）{//查询订单库pos.add(getPOrder());//等待barrier.await();}});T1.start();//查询派送订单Thread T2 = new Thread(() ->{while(存在未对账订单）{//查询订单库pos.add(getDOrder());//等待barrier.await();}});T2.start();
}

总结

CountDownLatch和CyclicBarrier是Java并发包提供的两个非常应用的线程同步工具类，这两个工具类的用法区别在这里还是有必要再强调一下的。CountDownLatch主要用于解决一个线程等待多个线程的场景可以类比于旅游团长要等待所有的旅客到齐才能去下一个景点，而CyclicBarrier是一组线程之间互相等待，更像是几个驴友之间的不离不弃。