并发工具类

JDK提供的CountDownLatch、CyclicBarrier、Semaphore用于并发流程控制。Exchanger工具则提供线程之间数据交换的手段。

1.CountDownLatch

构造方法指明计数器的大小，控制线程等待通知。

• countDown()：计数器减一
• await()：阻塞，直到计数器内部的值见到0才会被唤醒。
• await(long time)：设置超时时间的阻塞。

2.CyclicBarrier

构造方法指定屏障的大小n，仅当最后一个线程调用await后，所有被阻塞的线程组才会被同时释放。

• new CyclicBarrier(num,Runnable job)：构造方法，num个线程阻塞后，优先执行job方法。
• await()：调用线程进入阻塞状态。当前仅当第n个线程await后，前n-1和当前线程才会被唤醒。
• getNumberWaiting()：获取CyclicBarrier阻塞的线程数量。

CyclicBarrier可以重新reset计数器，线程重新执行任务。而CountDownLatch不能。

应用案例——汇总报账

多个线程分别统计每个sheet表的账目，所有线程执行完后，屏障释放，汇总线程执行。

public class BarrierUseCase implements Runnable{
    private CyclicBarrier barrier = new CyclicBarrier(4, this);
    private static ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
    private static Executor executors = Executors.newFixedThreadPool(4);
    public void executeTask() {
        for (int i = 0; i < 4; i++) {
            executors.execute(new Runnable() {
                @Override
                public void run() {
                    map.put(Thread.currentThread().getName(), 1);
                    try {
                        barrier.await();
                        System.out.println(Thread.currentThread().getName()+"线程关闭");
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    } catch (BrokenBarrierException e) {
                        throw new RuntimeException(e);
                    }
                }
            });
        }
    }
    
    @Override
    public void run() {
        int res = 0;
        for (Map.Entry<String, Integer> entry : map.entrySet()) {
            res += entry.getValue();
        }
        System.out.println("已经汇账完毕"+res);
    }
    
    public static void main(String[] args) {
        new BarrierUseCase().executeTask();
    }
}

3.Semaphore

控制可以同时访问共享资源的线程数量。构造函数时指定计数器数量大小。

• acquire()：阻塞式获取
• release()：释放信号量

public class SemaphoreTest {
    private static Semaphore s = new Semaphore(5);

    public static void main(String[] args) {
        Executor executor = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        s.acquire();
                        System.out.println(Thread.currentThread().getName() + "进行");
                        Thread.sleep(2000);
                        s.release();
                    } catch (Exception e) {
                    }
                }
            });
        }
    }
}

4.Exchanger

线程之间交换数据。当两个线程都执行到了同步点，则可以交换自己线程的数据给对方。

• exchange()：两个线程只有一个线程执行该方法，则会被一直等待。直到两个线程都执行了该方法。
• exchange(timeout)：超时最大等待时间。

应用案例——校对工作

public class ExchangeTest {
    static Exchanger<String> ex = new Exchanger<>();

    public static void main(String[] args) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    String str = ex.exchange(Thread.currentThread().getName() + "的数据");
                    //当前线程和传过来的数据str进行校对逻辑
                    System.out.println(str);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        },"thread01").start();
        
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(2000);
                    String str = ex.exchange(Thread.currentThread().getName() + "的数据");
                    System.out.println(str);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        },"thread02").start();
    }
}