Java 并发编程

Posted 2021-10-23 ybk-cherish

1.

同步器是一些使线程能够等待另一个线程的对象，允许它们协调动作。最常用的同步器是CountDownLatch和Semaphore，不常用的是Barrier 和Exchanger

CountDownLatch是一个同步辅助类，在完成一组正在其他线程中执行的操作前需要一个或多个其他的线程处于阻塞状态时我们可以使用它，如田径比赛中，将每一名运动员看做一个工作线程，成绩记录员看做另一线程，显然成绩记录员开始记录员开始记录成绩前需要所有的运动员都先做好比赛准备并且同时开始起跑，我们需要就最后参赛者所的成绩得出一个排名，显然这个排名也必须得等所有的参赛者都完成了比赛后才能计算得到。这样的场景正好可以用CountDownLatch来实现。

/** 
 *JAVA同步器之 CountDownLatch（不能循环使用，如果需要循环使用可以考虑使用CyclicBarrier） 
 *两种比较常规用法： 
 *1：new CountDownLatch(1);所有的线程在开始工作前需要做一些准备工作，当所有的线程都准备到位后再统一执行时有用 
 *2：new CountDownLatch(THREAD_COUNT);当所有的线程都执行完毕后，等待这些线程的其他线程才开始继续执行时有用 
 */  
public class CountDownLatchTest {  
  
    private static final int THREAD_COUNT=10;  
    //在调用startSingal.countDown()之前调用了startSingal.await()的线程一律等待，直到startSingal.countDown()的调用  
    private static final CountDownLatch startSingal=new CountDownLatch(1);  
    //在finishedSingal的初始化记数量通过调用finishedSingal.countDown()减少为0时调用了finishedSingal.await()的线程一直阻塞  
    private static final CountDownLatch finishedSingal=new CountDownLatch(THREAD_COUNT);  
    public static void main(String[] args) throws InterruptedException {  
        for(int i=0;i<THREAD_COUNT;i++){  
            new Thread("Task "+i){  
                public void run() {  
                    System.out.println(Thread.currentThread().getName()+" prepared!!");  
                    try {  
                        startSingal.await();  
                    } catch (InterruptedException e) {  
                        // TODO Auto-generated catch block  
                        e.printStackTrace();  
                    }  
                    //do something;  
                    System.out.println(Thread.currentThread().getName()+" finished!!");  
                    finishedSingal.countDown();  
                };  
            }.start();  
        }  
        Thread.sleep(1000);  
        startSingal.countDown();//所有的线程被唤醒，同时开始工作  
        finishedSingal.await();//等待所有的线程完成!!  
        System.out.println("All task are finished!!");  
    } 

Barrier使用示例：

/** 
 * JAVA同步器之Barrier(能够循环使用，当计数器增加到Barrier的初始化计数器之后马上会被置为0为下一次循环使用做准备) 
 * Barrier能够为指定的一个或多个（一般为多个）线程设置一道屏障，只有当所有的线程都到达该屏障后才能一起冲过该屏障继续其他任务 一般可以new 
 * CyclicBarrier(ThreadCount)来进行初始化，也可以new  CyclicBarrier(ThreadCount,RunableAction)当初始化数量的线程都调用 
 * 了await()方法后触发RunableAction线程，也可以通过初始化一个new CyclicBarrier(ThreadCount+1)的Barrier在前置线程未执行完成时一直阻塞一个或多个 
 * 后续线程，这一点类似于CountDownLatch 
 */  
public class BarrierTest {  
  
    private static final int THREAD_COUNT = 10;  
    private static final CyclicBarrier barrier = new CyclicBarrier(  
            THREAD_COUNT + 1, new Runnable() {  
  
                @Override  
                public void run() {  
                    System.out.println("All task are prepared or finished!!");  
                }  
            });  
  
    public static void main(String[] args) throws InterruptedException,  
            BrokenBarrierException {  
        for (int i = 0; i < THREAD_COUNT; i++) {  
            new Thread("Task " + i) {  
                public void run() {  
                    try {  
                        System.out.println(Thread.currentThread().getName()  
                                + " prepared!!");  
                        barrier.await();  
                    } catch (InterruptedException e) {  
                        // TODO Auto-generated catch block  
                        e.printStackTrace();  
                    } catch (BrokenBarrierException e) {  
                        // TODO Auto-generated catch block  
                        e.printStackTrace();  
                    }  
                    // do something  
                    System.out.println(Thread.currentThread().getName()  
                            + " finished!!");  
                };  
            }.start();  
        }  
        barrier.await();  
        // --------------开始准备循环使用--------------   
        for (int i = 0; i < THREAD_COUNT; i++) {  
            new Thread("Task " + i) {  
                public void run() {  
  
                    // do something  
                    System.out.println(Thread.currentThread().getName()  
                            + " finished!!");  
                    try {  
                        barrier.await();  
                    } catch (InterruptedException e) {  
                        // TODO Auto-generated catch block  
                        e.printStackTrace();  
                    } catch (BrokenBarrierException e) {  
                        // TODO Auto-generated catch block  
                        e.printStackTrace();  
                    }  
                };  
            }.start();  
        }  
        barrier.await();  
    } 

Exchanger使用示例：

/** 
 *JAVA同步器之Exchanger 
 *Exchanger 可以实现工作的线程间完成安全的数据交换 
 */  
public class ExchangerTest {  
  
    final static Exchanger<List<String>> exchanger = new Exchanger<List<String>>();  
  
    public static void main(String[] args) {  
        new Producer("Producer", exchanger).start();  
        new Consumer("Consumer", exchanger).start();  
    }  
  
    static class Producer extends Thread {  
        private Exchanger<List<String>> exchanger;  
  
        /** 
         *  
         */  
        public Producer(String threadName, Exchanger<List<String>> exchanger) {  
            super(threadName);  
            this.exchanger = exchanger;  
        }  
  
        /* 
         * (non-Javadoc) 
         *  
         * @see java.lang.Thread#run() 
         */  
        @Override  
        public void run() {  
            List<String> products = new ArrayList<String>();  
            for (int i = 0; i < 10; i++) {  
                products.add("product " + i);  
            }  
            try {  
                List<String> results = exchanger.exchange(products);  
                System.out.println("get results from consumer");  
                for (String s : results) {  
                    System.out.println(s);  
                }  
            } catch (InterruptedException e) {  
                // TODO Auto-generated catch block  
                e.printStackTrace();  
            }  
        }  
    }  
  
    static class Consumer extends Thread {  
        private Exchanger<List<String>> exchanger;  
  
        /** 
         *  
         */  
        public Consumer(String threadName, Exchanger<List<String>> exchanger) {  
            super(threadName);  
            this.exchanger = exchanger;  
        }  
  
        /* 
         * (non-Javadoc) 
         *  
         * @see java.lang.Thread#run() 
         */  
        @Override  
        public void run() {  
            List<String> products = new ArrayList<String>();  
            for (int i = 0; i < 10; i++) {  
                products.add("consumed " + i);  
            }  
            try {  
                List<String> results = exchanger.exchange(products);  
                System.out.println("got products from produces");  
                for (String s : results) {  
                    System.out.println(s);  
                }  
            } catch (InterruptedException e) {  
                // TODO Auto-generated catch block  
                e.printStackTrace();  
            }  
        }  
    }

---------------------
作者：soliderzzz
来源：CSDN
原文：https://blog.csdn.net/qq_18975791/article/details/80728045