温习 Java 常见的并发工具类

水深无声 2022-05-28 06:57 161阅读 0赞

对着书敲一遍，温故而知新

**CountDownLatch**，**CyclicBarrier**，**Semaphore**，**Exchanger**

*  CountDownLatch

package com.wen.java.concurrent.tools;
    
    import java.util.concurrent.CountDownLatch;
    
    /*CountDownLatch的第一次交互是主线程等待其他线程。主线程必须在启动其他线程后立即调用CountDownLatch.await()方法。 这样主线程的操作就会在这个方法上阻塞，直到其他线程完成各自的任务。*/
    public class CountDownLatchDemo {
    
        static CountDownLatch c = new CountDownLatch(2);
    
        public static void main(String[] args) throws Exception {
    
            new Thread(new Runnable(){
                public void run() {
                    System.out.println("thread 1");
                    c.countDown();
                    System.out.println("thread 2");
                    c.countDown();
                }
            }).start();;
            c.await();
            System.out.println("main 3");
    
        }
    
    }

*  CyclicBarrier

package com.wen.java.concurrent.tools;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /* * CyclicBarrier 的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞， * 直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活。CyclicBarrier默认的构造方法是CyclicBarrier(int parties)， * 其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。 */
    //赛马
    class Horse implements Runnable {
        //赛马的编号
        private static int counter = 0;
        private final int id = counter++;
        //跑过的步数
        private int strides = 0;
        private static Random rand = new Random(47);
        private static CyclicBarrier barrier;
        public Horse(CyclicBarrier barrier) {
       this.barrier = barrier;}
        public synchronized int getStrides() {
       return strides;}
        public void run() {
            try{
                //当现成获得运行权限时
                while(!Thread.interrupted()) {
                    synchronized(this) {
                        strides += rand.nextInt(10);
                    }
                    //当前线程达到屏障点
                    barrier.await();
                }
            }catch(InterruptedException e) {
    
            }catch(BrokenBarrierException e) {
                throw new RuntimeException(e);
            }
        }
        public String toString(){
       return "Horse " + id + "";}
        //用*号打印出赛马跑的轨迹
        public String tracks() {
            StringBuilder s = new StringBuilder();
            for(int i = 0; i < getStrides(); i++) {
                s.append("*");
            }
            s.append(id);
            return s.toString();
        }
    }
    
    class HorseRace{
        static final int FINISH_LINE = 75;
        private List<Horse> horses = new ArrayList<Horse>();
        private ExecutorService exec = Executors.newCachedThreadPool();
        private CyclicBarrier barrier;
        public HorseRace(int nHorses, final int pause) {
            //初始化barrier
            barrier = new CyclicBarrier(nHorses, new Runnable(){
                public void run(){
                    StringBuilder s = new StringBuilder();
                    for(int i = 0; i < FINISH_LINE; i++) {
                        s.append("=");
                    }
                    System.out.println(s);
                    for(Horse horse : horses) {
                        System.out.println(horse.tracks());
                    }
                    //结束循环的条件
                    for(Horse horse : horses) {
                        if(horse.getStrides() >= FINISH_LINE) {
                            System.out.println(horse + " won!");
                            exec.shutdownNow();
                            return;
                        }
                    }
                    try{
                        TimeUnit.MILLISECONDS.sleep(pause);
                    }catch(InterruptedException e) {
                        System.out.println("barrier-action sleep interrupted");
                    }
                }
            });
            for(int i = 0; i < nHorses; i++) {
                Horse horse = new Horse(barrier);
                horses.add(horse);
                exec.execute(horse);
            }
        }
        public static void main(String[] args) {
            int nHorses = 7;
            int pause = 200;
            new HorseRace(nHorses, pause);
        }
    }

*  Semaphore

package com.wen.java.concurrent.tools;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Semaphore;
    import java.util.concurrent.TimeUnit;
    
    /* * Semaphore（信号量）是用来控制同时访问特定资源的线程数量，它通过协调各个线程， * 以保证合理的使用公共资源 * 比如数据库连接。假如有一个需求，要读取几万个文件的数据，因为都是IO密集型任务， * 我们可以启动几十个线程并发的读取，但是如果读到内存后，还需要存储到数据库中， * 而数据库的连接数只有10个，这时我们必须控制只有十个线程同时获取数据库连接保存数据， * 否则会报错无法获取数据库连接。这个时候，我们就可以使用Semaphore来做流控 */
    public class SemaphoreDemo { 
    
        private static final int THREAD_COUNT = 30;
    
        private static int counter = 0;
    
        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(new Runnable() {
                    public void run(){
                        try{
                            s.acquire();
                            System.out.println("save data" + ++counter);
                            //模拟耗时操作
                            TimeUnit.SECONDS.sleep(2);
                            s.release();
                        }catch(InterruptedException e) {
    
                        }
                    }
                });
            }
            threadPool.shutdown();
        }
    
    }

*  Exchanger

package com.wen.java.concurrent.tools;
    
    import java.util.concurrent.Exchanger;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /* * Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。 * 它提供一个同步点，在这个同步点两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据， * 如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange，当两个线程都到达同步点时， * 这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。 */
    public class ExchangerDemo { 
    
        private static final Exchanger<String> exgr = new Exchanger<String>();
    
        private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
    
        public static void main(String[] args) {
            threadPool.execute(new Runnable() {
                public void run(){
                    try{
                        String A = "银行流水A";
                        exgr.exchange(A);
                    }catch(InterruptedException e){}
                }
            });
            threadPool.execute(new Runnable() {
                public void run(){
                    try{
                        String B = "银行流水B";
                        String A = exgr.exchange(B);
                        System.out.println(A.equals(B) + " A is: " + A + " B is: " + B);
                    }catch(InterruptedException e){}
                }
            });
            threadPool.shutdown();
        }
    
    }