前言
往期文章在热点数据如何更新的一篇文章中有提到对账系统。其实我在实际业务场景中是有遇到过类似对账的优化问题的。说优化之前要掌握一点就是一定要掌握Java并发包的相关特性。本章节对此有很大依赖。
- 热点数据高效更新文章：
inventory hint,解决热点数据如何高效更新
Java并发包简说
CountDownLatch和CyclicBarrier
区别
CountDownLatch:
- 不可以重复使用，计数器无法被重置
经典案例比如门卫休息休要等所有人下班才可以关门休息
CyclicBarrier：
- 可以重复使用，是一个同步辅助类，允许一组线程相互等待，直到到达某个公共屏障点(common barrier point)。假设设计一组固定大小的线程的程序中，这些线程必须不是的互相等待，此时就可以使用CyclicBarrier。因为该barrier在释放等待线程后可以重用，从而称之为循环的barrier。
经典案例：比如运动员跑步，需要所有人准备好之后裁判才可以发令让大家在同一时刻去跑。有依赖关系
案例
有一天，老大匆忙赶来，提到对账系统最近变得越来越缓慢，希望能迅速进行优化。经过了解对账系统的业务流程，发现其实相当简单：用户通过在线商城下单会生成电子订单并存储在订单数据库中；随后物流会生成派送单用于发货，派送单则保存在派送单库中。为了避免漏发或重复派送，对账系统每天会核查是否存在异常订单。
目前对账系统的处理逻辑很简单：首先查询订单，然后查询派送单，接着比对订单和派送单，将差异记录写入差异库。对账系统的核心代码经过抽象后，也并不复杂，主要是在单线程中循环执行订单和派送单的查询，进行对账操作，最后将结果写入差异库。
伪代码
代码语言：java复制while(存在未对账订单){
// 查询未对账订单
pos = getPOrders();
// 查询派送单
dos = getDOrders();
// 执行对账操作
diff = check(pos, dos);
// 差异写入差异库
save(diff);
} 利用Java并行优化对账系统
首先要找出对账系统的瓶颈所在。目前，由于订单量和派送单量庞大，导致查询未对账订单 getPOrders() 和查询派送单 getDOrders() 的速度较慢。是否有一种快速优化的方法呢？目前对账系统是单线程执行的。对于这样的串行系统，优化性能的第一个想法是能否利用多线程并行处理。
因此，我们可以看出对账系统的瓶颈在哪里：查询未对账订单 getPOrders() 和查询派送单 getDOrders() 是否能够并行处理呢？很显然，这两个操作之间并没有依赖关系。将这两个耗时操作并行化后，与单线程执行相比，您会发现在相同时间段内，并行执行的吞吐量接近单线程的两倍，优化效果颇为明显。
有了这个思路，接下来我们看看如何用代码实现。在下面的代码中，我们创建了两个线程 T1 和 T2，分别并行执行查询未对账订单 getPOrders() 和查询派送单 getDOrders() 的操作。主线程则负责执行对账操作 check()和将差异写入 save() 的操作。值得注意的是：主线程需要等待线程 T1 和 T2 执行完毕后才能执行 check() 和 save() 这两个操作。为此，我们通过调用 T1.join() 和 T2.join() 实现等待，当线程 T1 和 T2 结束时，调用了 T1.join() 和 T2.join() 的主线程将从阻塞状态中解除，随后执行后续的 check() 和 save() 操作。
伪代码
代码语言：java复制while(存在未对账订单){
// 查询未对账订单
Thread T1 = new Thread(()->{
pos = getPOrders();
});
T1.start();
// 查询派送单
Thread T2 = new Thread(()->{
dos = getDOrders();
});
T2.start();
// 等待 T1、T2 结束
T1.join();
T2.join();
// 执行对账操作
diff = check(pos, dos);
// 差异写入差异库
save(diff);
} 用 CountDownLatch 实现线程等待
经过上述优化，基本上可以向老板报告工作完成了，但仍有一些遗憾之处。我相信您也已经注意到了，在 while 循环中每次都会创建新的线程，而创建线程是一个耗时的操作。因此，最好能够重复利用已创建的线程。您想到了线程池，确实，线程池能够解决这个问题。
通过线程池进行优化后：我们首先创建了一个固定大小为2的线程池，并在 while 循环中重复利用这些线程。一切看起来都进行得很顺利，但似乎有一个问题无法解决，即主线程如何知道 getPOrders() 和 getDOrders() 这两个操作何时执行完成。在前面的方案中，主线程通过调用线程 T1 和 T2 的 join() 方法来等待它们退出，但是在线程池方案中，线程根本就不会退出，因此 join() 方法失效了。
那么，如何解决这个问题呢？您可以想出许多方法，其中最直接的方法是使用一个计数器。将其初始值设为2，执行完 pos = getPOrders(); 后减 1，执行完 dos = getDOrders(); 后也减 1。主线程在这之后等待计数器等于0；当计数器等于0时，说明这两个查询操作已执行完毕。等待计数器为0实际上是一种条件变量，使用管程实现起来也并不复杂。
然而，我并不建议在实际项目中实施上述方案，因为Java并发包中已经提供了类似功能的工具类：CountDownLatch，我们直接使用即可。在下面的代码示例中，我们在 while 循环中首先创建了一个CountDownLatch，计数器的初始值为2。在 pos = getPOrders(); 和 dos = getDOrders(); 两个语句后，通过调用 latch.countDown(); 实现对计数器的减1操作。在主线程中，通过调用 latch.await(); 实现对计数器等于0的等待。
伪代码
代码语言：java复制// 创建 2 个线程的线程池
Executor executor =
Executors.newFixedThreadPool(2);
while(存在未对账订单){
// 计数器初始化为 2
CountDownLatch latch =
new CountDownLatch(2);
// 查询未对账订单
executor.execute(()-> {
pos = getPOrders();
latch.countDown();
});
// 查询派送单
executor.execute(()-> {
dos = getDOrders();
latch.countDown();
});
// 等待两个查询操作结束
latch.await();
// 执行对账操作
diff = check(pos, dos);
// 差异写入差异库
save(diff);
}进一步优化性能
经过以上一系列的优化，终于可以松一口气，准备交付项目。然而，在交付之前，再次审视一番是值得的，或许还存在优化的空间。
前面我们已经实现了将 getPOrders() 和 getDOrders() 这两个查询操作并行化，但是这两个查询操作与对账操作 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需要等待。当计数器等于0时，通知线程 T3，唤醒等待的线程 T1 或 T2，并将计数器重置为2。如此，线程 T1 和 T2 在生产下一条数据时，可以继续使用这个计数器。
建议不要在实际项目中直接实现这一逻辑，因为Java并发包中已经提供了相关的工具类：CyclicBarrier。在下面的代码中，我们首先创建了一个初始值为2的CyclicBarrier计数器。需要注意的是，在创建CyclicBarrier时，传入了一个回调函数。当计数器减至0时，该回调函数会被调用。
线程 T1 负责查询订单，每查到一条数据，调用barrier.await()将计数器减1，并等待计数器变为0。线程 T2 负责查询派送单，处理方式与线程 T1 类似。当 T1 和 T2 都调用barrier.await()时，计数器会减至0，此时 T1 和 T2可以继续执行下一步操作，并调用barrier的回调函数执行对账操作。
值得一提的是，CyclicBarrier的计数器具有自动重置功能。当计数器减至0时，会自动重新设定为您设置的初始值。这一特性确实方便实用。
伪代码
代码语言：java复制// 订单队列
Vector<P> pos;
// 派送单队列
Vector<D> dos;
// 执行回调的线程池
Executor executor =
Executors.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(getPOrders());
// 等待
barrier.await();
}
});
T1.start();
// 循环查询运单库
Thread T2 = new Thread(()->{
while(存在未对账订单){
// 查询运单库
dos.add(getDOrders());
// 等待
barrier.await();
}
});
T2.start();
}CountDownLatch 和 CyclicBarrier 是Java并发包提供的两个非常便捷的线程同步工具类。在这里，有必要再次强调它们之间的区别：CountDownLatch 主要用于解决一个线程等待多个线程的情况，可以类比于旅游团团长必须等待所有游客齐集后才能继续前行；而CyclicBarrier 则是一组线程相互等待，有点像几个驴友之间的互助合作。此外，CountDownLatch 的计数器不支持重复利用，即一旦计数器降至0，后续调用await()的线程将直接通过。相比之下，CyclicBarrier 的计数器可以循环利用，同时具有自动重置功能，一旦计数器减至0，将会自动重置为设定的初始值。此外，CyclicBarrier 还支持设置回调函数，功能更加丰富。
我正在参与2024腾讯技术创作特训营最新征文，快来和我瓜分大奖！
