CyclicBarrier CountDownLatch Semaphore Exchanger

我不是女神ヾ 2022-06-03 00:56 250阅读 0赞

## 同步屏障CyclicBarrier ##

> CyclicBarrier 的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活。CyclicBarrier默认的构造方法是CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。

public class Test { 
    
        private static final int THREAD_NUM = 5;
    
        public static class WorkerThread implements Runnable { 
    
            CyclicBarrier cyclicBarrier;
    
            private WorkerThread(CyclicBarrier cyclicBarrier) {
                this.cyclicBarrier = cyclicBarrier;
            }
    
            @Override
            public void run() {
                try {
                    System.out.println("Worker's waiting");
                    cyclicBarrier.await();
                    System.out.println("ID:" + Thread.currentThread().getId() + " Working");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) {
            ExecutorService pool = Executors.newCachedThreadPool();
            CyclicBarrier cyclicBarrier = new CyclicBarrier(THREAD_NUM, () -> System.out.println("Inside Barrier"));
            for (int i = 0; i < THREAD_NUM; i++) {
                pool.execute(new WorkerThread(cyclicBarrier));
            }
            pool.shutdown();
        }
    
    }
    /* 以下是输出： Worker's waiting Worker's waiting Worker's waiting Worker's waiting Worker's waiting Inside Barrier ID:15 Working ID:12 Working ID:13 Working ID:11 Working ID:14 Working */

## CountDownLatch ##

> CountDownLatch 允许一个或多个线程等待其他线程完成操作。

public class Test { 
    
        static CountDownLatch c = new CountDownLatch(2);
    
        public static void main(String[] args) {
            new Thread(() -> { System.out.println(1); c.countDown(); System.out.println(2); c.countDown(); }).start(); try { c.await(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("3"); } }

## Semaphore ##

> Semaphore（信号量）是用来控制同时访问特定资源的线程数量，它通过协调各个线程，以保证合理的使用公共资源。很多年以来，我都觉得从字面上很难理解Semaphore所表达的含义，只能把它比作是控制流量的红绿灯，比如XX马路要限制流量，只允许同时有一百辆车在这条路上行使，其他的都必须在路口等待，所以前一百辆车会看到绿灯，可以开进这条马路，后面的车会看到红灯，不能驶入XX马路，但是如果前一百辆中有五辆车已经离开了XX马路，那么后面就允许有5辆车驶入马路，这个例子里说的车就是线程，驶入马路就表示线程在执行，离开马路就表示线程执行完成，看见红灯就表示线程被阻塞，不能执行。

public class Test {
    
        private static final int THREAD_COUNT = 30;
    
        private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
        private static Semaphore s = new Semaphore(10);
    
        public static void main(String[] args) {
            for (int i = 0; i < THREAD_COUNT; i++) {
                threadPool.execute(() -> {
                    try {
                        s.acquire();
                        TimeUnit.SECONDS.sleep(1);
                        System.out.println("save data");
                        s.release();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
    
                });
            }
            threadPool.shutdown();
    
        }
    
    }

## Exchanger ##

> Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点，在这个同步点两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据， 如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。

public class Test {
    
        private static final Exchanger<String> exgr = new Exchanger<String>();
        private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
    
        public static void main(String[] args) {
            threadPool.execute(()->{
                try {
                    String A = "银行流水A";// A录入银行流水数据
                    String B = exgr.exchange(A);
                    System.out.println(B);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
            });
            threadPool.execute(()->{
                try {
                    String B = "银行流水B";// B录入银行流水数据
                    String A = exgr.exchange("B");
                    System.out.println("A和B数据是否一致：" + A.equals(B) + ",A录入的是：" + A + ",B录入是：" + B);
    
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
            });
        }
    }