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

系统管理员 2020-11-07 02:09 607阅读 0赞

## 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.  booleanhasQueuedThreads()：是否有线程正在等待获取许可证。
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();
        }
    }
````