一 点睛

若干个线程并发执行某个特定的任务，然后等到所有的子任务都执行结束后在统一汇总，比如用户想要查询自己三年以来银行账户的流水，为了保证运行的数据库的数据量在一个恒定的范围之内，通常数据只会保存一年的记录，其他历史记录或被备份到磁盘，或者被存储到 hive 数据仓库，或者被转至备份数据库之中，总之，想要三年的流水记录，需要若干次的查询最后汇总返回给用户，很明显这样的操作性能低下，用户体验差，如果我们将每一个渠道的查询交给一个线程或者若干干线程去查询，然后统一汇总，那么性能会提高很多，响应时间也会缩短不少。

Latch 设计模式指定了一个屏障，只有所有条件都达到满足的时候，门栓才能打开。

二 实战

1 Latch

package concurrent.latch;

public abstract class Latch {
    // 用于控制多少个线程完成任务时才能打开阀门
    protected int limit;

    // 通过构造函数传入 limit
    public Latch(int limit) {
        this.limit = limit;
    }

    // 该方法会使得当前线程一直等待，直到所有的线程都完成工作，被阻塞的线程是允许被中断的
    public abstract void await() throws InterruptedException;

    // 当任务线程完成工作之后调用该方法使得计数器减一
    public abstract void countDown();

    // 获取当前还有多少个线程没有完成任务
    public abstract int getUnarriveed();
}

2 CountDownLatch

package concurrent.latch;

/**
* @className: CountDownLatch
* @description: 无限等待门栓实现
* @date: 2022/4/25
* @author: cakin
*/
public class CountDownLatch extends Latch {

    public CountDownLatch(int limit) {
        super(limit);
    }

    @Override
    public void await() throws InterruptedException {
        synchronized (this) {
            // 当 limit > 0 时，当前线程进入阻塞状态
            while (limit > 0) {
                this.wait();
            }
        }
    }

    @Override
    public void countDown() {
        synchronized (this) {
            if (limit <= 0) {
                throw new IllegalStateException("all of task already arrived");
            }
            // 使 limit 减一，并且通知阻塞线程
            limit--;
            this.notifyAll();
        }
    }

    @Override
    public int getUnarriveed() {
        // 返回有多少线程还未完成任务
        return limit;
    }
}

3 ProgrammerTravel

package concurrent.latch;

import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/**
* @className: ProgrammerTravel
* @description: 程序员旅游线程
* @date: 2022/4/25
* @author: cakin
*/
public class ProgrammerTravel extends Thread {
    // 门栓
    private final Latch latch;
    // 程序员
    private final String programmer;
    // 交通工具
    private final String transportation;

    public ProgrammerTravel(Latch latch, String programmer, String transportation) {
        this.latch = latch;
        this.programmer = programmer;
        this.transportation = transportation;
    }

    @Override
    public void run() {
        System.out.println(programmer + " start take the transportation [" + transportation + "]");
        try {
            TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(10));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(programmer + " arrived by" + transportation);
        // 完成任务时使计数器减一
        latch.countDown();
    }
}

4 Test

package concurrent.latch;

public class Test {
    public static void main(String[] args) throws InterruptedException {
        // 定义 Latch，limit 为 4
        Latch latch = new CountDownLatch(4);
        new ProgrammerTravel(latch, "Alex", "Bus").start();
        new ProgrammerTravel(latch, "Gavin", "Walking").start();
        new ProgrammerTravel(latch, "Jack", "Subway").start();
        new ProgrammerTravel(latch, "Dillon", "Bicycle").start();
        // 当前线程（main 线程会进入阻塞，直到四个程序员全部都达到目的地）
        latch.await();
        System.out.println("== all of programmer arrvied ==");
    }
}

三 测试结果

Alex start take the transportation [Bus]

Dillon start take the transportation [Bicycle]

Jack start take the transportation [Subway]

Gavin start take the transportation [Walking]

Jack arrived bySubway

Gavin arrived byWalking

Dillon arrived byBicycle

Alex arrived byBus

== all of programmer arrvied ==