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

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();
        }
    }
}
原文地址:https://www.cnblogs.com/jmcui/p/11511413.html