多线程编程学习九(并发工具类).

CountDownLatch

1. CountDownLatch 允许一个或多个线程等待其他线程完成操作。
2. CountDownLatch 可以替代 join 的作用，并提供了更丰富的用法。
3. CountDownLatch 的 countDown 方法，N 会减1；CountDownLatch 的 await 方法会阻塞当前线程，直到 N 变成零。
4. CountDownLatch 不可能重新初始化或者修改 CountDownLatch 对象的内部计数器的值。
5. CountDownLatch 内部由 AQS 共享锁实现。

public class CountDownLatchTest {

    private static final CountDownLatch DOWN_LATCH = new CountDownLatch(2);

    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> {
            System.out.println(1);
            DOWN_LATCH.countDown();
            System.out.println(2);
            DOWN_LATCH.countDown();

        }).start();
        DOWN_LATCH.await();
        System.out.println("3");
    }
}

CyclicBarrier

1. CyclicBarrier 设置一个屏障（也可以叫同步点），拦截阻塞一组线程，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。
2. CyclicBarrier 默认的构造方法是CyclicBarrier（int parties），其参数表示屏障拦截的线程数量，每个线程调用await方法告诉 CyclicBarrier 我已经到达了屏障，然后当前线程被阻塞。
3. CyclicBarrier 还提供一个更高级的构造函数 CyclicBarrier（int parties，Runnable barrierAction），用于在线程到达屏障时，优先执行 barrierAction，方便处理更复杂的业务场景。
4. getNumberWaiting 方法可以获得 CyclicBarrier 阻塞的线程数量；isBroken()方法用来了解阻塞的线程是否被中断。
5. CyclicBarrier 的计数器可以使用 reset() 方法重置（CountDownLatch 的计数器只能使用一次）。所以 CyclicBarrier 能处理更为复杂的业务场景。例如，如果计算发生错误，可以重置计数器，并让线程重新执行一次。
6. CyclicBarrier 可以用于多线程计算数据，最后合并计算结果的场景。
7. CyclicBarrier 内部采用重入锁 ReentrantLock 实现。

public class BankWaterService implements Runnable {
 
    // 创建4个屏障，处理完之后执行当前类的run方法
    private CyclicBarrier barrier = new CyclicBarrier(4, this);
    // 假设有4个计算任务，所以只启动4个线程
    private Executor executor = Executors.newFixedThreadPool(4);
    // 保存每个任务的计算结果
    private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new ConcurrentHashMap<>();

    private AtomicInteger atomicInteger = new AtomicInteger(1);

    private void count() {
        for (int i = 0; i < 4; i++) {
            Thread thread = new Thread(() -> {
                // 当前任务的计算结果，计算过程忽略
                sheetBankWaterCount.put(Thread.currentThread().getName(), 1);
                // 计算完成，插入一个屏障
                try {
                    barrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }

            }, "线程" + atomicInteger.getAndIncrement());
            executor.execute(thread);
        }
    }

    @Override
    public void run() {
        int result = 0;
        // 汇总每个任务计算出的结果
        for (Map.Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()) {
            result += sheet.getValue();
        }
        //将结果输出
        sheetBankWaterCount.put("result", result);
        System.out.println(result);
    }

    public static void main(String[] args) {
        BankWaterService bankWaterCount = new BankWaterService();
        bankWaterCount.count();
    }
}

Semaphore

1. Semaphore（信号量）是用来控制同时访问特定资源的线程数量，它通过协调各个线程，以保证合理的使用公共资源。
2. Semaphore 可以用于做流量控制，特别是公用资源有限的应用场景，比如数据库连接。
3. Semaphore的构造方法 Semaphore(int permits) 接受一个整型的数字，表示可用的许可证数量。
4. 首先线程使用 Semaphore 的 acquire() 方法获取一个许可证，使用完之后调用 release() 方法归还许可证。还可以用 tryAcquire() 方法尝试获取许可证。
5. intavailablePermits()：返回此信号量中当前可用的许可证数。
6. intgetQueueLength()：返回正在等待获取许可证的线程数。
7. hasQueuedThreads()：是否有线程正在等待获取许可证。
8. Semaphore 内部使用 AQS 共享锁实现。

public class SemaphoreTest {

    private static final int THREAD_COUNT = 30;
    private static ExecutorService EXECUTOR = Executors.newFixedThreadPool(THREAD_COUNT);
    private static Semaphore SEMAPHORE = new Semaphore(10);
    private static AtomicInteger ATOMICINTEGER = new AtomicInteger(1);

    public static void main(String[] args) {
        for (int i = 0; i < THREAD_COUNT; i++) {
            EXECUTOR.execute(() -> {
                try {
                    SEMAPHORE.acquire();
                    System.out.println("save data" + ATOMICINTEGER.getAndIncrement());
                    SEMAPHORE.release();
                } catch (InterruptedException e) {
                }

            });
        }
        EXECUTOR.shutdown();
    }
}

Exchanger

1. Exchanger（交换者）是一个用于线程间协作的工具类 —— 用于线程间的数据交换。它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。这两个线程通过 exchange 方法交换数据，如果第一个线程先执行 exchange() 方法，它会一直等待第二个线程也执行 exchange 方法。
2. 可简单地将 Exchanger 对象理解为一个包含两个格子的容器，通过 exchanger 方法可以向两个格子中填充信息。当两个格子中的均被填充时，该对象会自动将两个格子的信息交换，然后返回给线程，从而实现两个线程的信息交换。
3. Exchanger 可用于遗传算法。（遗传算法：需要选出两个人作为交配对象，这时候会交换两人的数据，并使用交叉规则得出交配结果）
4. Exchanger 可用于校对工作，比如一份数据需要两个人同时进行校对，都校对无误后，才能进行后续处理。这时，就可以使用 Exchanger 比较两份校对结果。
5. Exchanger 内部采用无锁 CAS 实现，Exchange 使用了内部对象 Node 的两个属性 — item 、match，分布存储两个线程的值。

public class ExchangerTest {

    private static final Exchanger<String> exchange = new Exchanger<>();
    private static ExecutorService threadPool = Executors.newFixedThreadPool(2);

    public static void main(String[] args) {
        threadPool.execute(() -> {
            try {
                String result = exchange.exchange("数据A");
                System.out.println("A的exchange结果：" + result);
            } catch (InterruptedException e) {
            }

        });
        threadPool.execute(() -> {
            try {
                String result = exchange.exchange("数据B");
                System.out.println("B的exchange结果：" + result);
            } catch (InterruptedException e) {
            }
        });
        threadPool.shutdown();
    }
}

Phaser

多阶段栅栏，可以在初始时设定参与线程数，也可以中途注册/注销参与者，当到达的参与者数量满足栅栏设定的数量后，会进行阶段升级（advance）。以下是 Phaser 的一些基础概念。

1. phase(阶段)：在任意时间点，Phaser 只处于某一个 phase(阶段)，初始阶段为 0，最大达到 Integerr.MAX_VALUE，然后再次归零。当所有 parties 参与者都到达后，phase 值会递增。
2. parties(参与者)：指参与线程。Phaser 既可以在初始构造时指定参与者的数量，也可以中途通过register、bulkRegister、arriveAndDeregister等方法注册/注销参与者。
3. arrive(到达)：Phaser 注册完 parties（参与者）之后，参与者的初始状态是 unarrived 的，当参与者到达（arrive）当前阶段（phase）后，状态就会变成 arrived。
4. advance(进阶)：当阶段的到达参与者数满足条件后（注册的数量等于到达的数量），阶段就会发生进阶（advance）—— 也就是phase值+1。
5. termination（终止）：代表当前 Phaser 对象达到终止状态。
6. Tiering（分层）：一种树形结构，通过构造函数可以指定当前待构造的 Phaser对象的父结点。
   6.1 之所以引入 Tiering，是因为当一个 Phaser 有大量参与者（parties）的时候，内部的同步操作会使性能急剧下降，而分层可以降低竞争，从而减小因同步导致的额外开销。
   6.2 在一个分层 Phasers 的树结构中，注册和撤销子Phaser或父Phaser是自动被管理的。当一个Phaser的参与者（parties）数量变成 0 时，如果有该 Phaser 有父结点，就会将它从父结点中溢移除。

public class PhaserTest1 implements Runnable {

    private final Phaser phaser;

    public PhaserTest1(Phaser phaser) {
        this.phaser = phaser;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + ": 开始执行任务，当前phase =" + phaser.getPhase() + "");
        // 1.等待其他参与者线程到达
        // 2.不响应中断，也就是说即使当前线程被中断，arriveAndAwaitAdvance方法也不会返回或抛出异常，而是继续等待。如果希望能够响应中断，可以参考 awaitAdvanceInterruptibly 方法。
        int advance = phaser.arriveAndAwaitAdvance();
        System.out.println(Thread.currentThread().getName() + ": 结束执行任务，当前phase =" + advance + "");
    }

    public static void main(String[] args) {
        Phaser phaser = new Phaser();
        for (int i = 0; i < 10; i++) {
            // 新增参与者数量
            phaser.register();
            new Thread(new PhaserTest1(phaser), "Thread-" + i).start();
        }
    }
}