java多线程 CountDownLatch、CyclicBarrier和Semaphore

桃扇骨 2022-12-28 07:48 230阅读 0赞

## java多线程 CountDownLatch、CyclicBarrier和Semaphore ##

### 1. 概念 ###

CountDownLatch：减数计数器，只有当所有线程到达才会往下执行。  
CyclicBarrier：循环阑珊，所有线程到达才会往下执行，与countDownLatch需要手动countDown自减，CyclicBarrier不需要，而且可以重置循环使用。  
Semaphore：可以用来控制同时访问特定资源的线程数量，一般用于限流，控制访问量，比如停车场。

### 2.案例 ###

**线程工具类准备**

import com.google.common.util.concurrent.ThreadFactoryBuilder;
    
    import java.util.concurrent.*;
    
    /** * 线程池常量 * * @author 20023262 * @date 2020/12/15 * @since 1.0 */
    public class ThreadConstant { 
    
        /** * 核心线程数 */
        public static final int corePoolSize = 30;
    
        /** * 最大线程数 */
        public static final int maximumPoolSize = 200;
    
        /** * 线程存活时间 */
        public static final long keepAliveTime = 0L;
    
        /** * 时间单位 */
        public static final TimeUnit milliseconds = TimeUnit.MILLISECONDS;
    
        /** * 线程队列 */
        public static final BlockingQueue<Runnable> linkedBlockingDeque = new LinkedBlockingDeque<>(1024);
    
        /** * 线程名称 */
        public static final ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-pool-%d").build();
    
        /** * 拒绝策略：1.默认抛出异常AbortPolicy；2.DiscardPolicy拒绝任务不抛异常，3.DiscardOldestPolicy弹出最先的添加进去 */
        public static final RejectedExecutionHandler abortPolicy = new ThreadPoolExecutor.AbortPolicy();
        public static final RejectedExecutionHandler discardPolicy = new ThreadPoolExecutor.DiscardPolicy();
        public static final RejectedExecutionHandler discardOldestPolicy = new ThreadPoolExecutor.DiscardOldestPolicy();
    
    }

import java.util.concurrent.*;
    
    /** * 线程池工具类 * * @author 20023262 * @date 2020/12/15 * @since 1.0 */
    public class ThreadPoolUtil { 
    
        /** * 创建线程池，默认抛出异常AbortPolicy */
        public static ExecutorService abortPolicyPool = new ThreadPoolExecutor(
                ThreadConstant.corePoolSize,
                ThreadConstant.maximumPoolSize,
                ThreadConstant.keepAliveTime,
                ThreadConstant.milliseconds,
                ThreadConstant.linkedBlockingDeque,
                ThreadConstant.namedThreadFactory,
                ThreadConstant.abortPolicy
        );
    
        /** * 创建线程池，DiscardPolicy拒绝任务不抛异常 */
        public static ExecutorService discardPolicyPool = new ThreadPoolExecutor(
                ThreadConstant.corePoolSize,
                ThreadConstant.maximumPoolSize,
                ThreadConstant.keepAliveTime,
                ThreadConstant.milliseconds,
                ThreadConstant.linkedBlockingDeque,
                ThreadConstant.namedThreadFactory,
                ThreadConstant.discardPolicy
        );
    
        /** * 创建线程池，DiscardOldestPolicy弹出最先的添加进去 */
        public static ExecutorService discardOldestPolicyPool = new ThreadPoolExecutor(
                ThreadConstant.corePoolSize,
                ThreadConstant.maximumPoolSize,
                ThreadConstant.keepAliveTime,
                ThreadConstant.milliseconds,
                ThreadConstant.linkedBlockingDeque,
                ThreadConstant.namedThreadFactory,
                ThreadConstant.discardOldestPolicy
        );
    
    }

**CountDownLatch**

import java.util.concurrent.CountDownLatch;
    
    /** * 线程类 * * @author 20023262 * @date 2020/12/15 * @since 1.0 */
    public class CountDownLatchDemo implements Runnable { 
    
        private CountDownLatch countDownLatch;
    
        public CountDownLatchDemo(CountDownLatch countDownLatch) { 
            this.countDownLatch = countDownLatch;
        }
    
        /** * 这里存在线程安全问题，不加锁，countDownLatch.countDown();自减的时候会出现问题 * 当并发100个线程进来的时候，countDownLatch最初是100，第一个还没有来得及countDown，第二个已经完成，此时countDownLatch是99 * 但是第一个取得值依然是100，这就会导致重复自减 */
        @Override
        public void run() { 
            synchronized (countDownLatch) { 
                /**执行子线程代码**/
                countDownLatch.countDown();
                System.out.println("thread counts = " + (countDownLatch.getCount()));
            }
    
        }
    }

import org.junit.Test;
    
    import java.util.concurrent.CountDownLatch;
    
    /** * CountDownLatch测试类 * 两个方法适用于两种不同的场景，countDownLatchTest2多用于批量处理相互之间没有依赖关系的相同任务 * countDownLatchTest适用于批量处理相互之间没有依赖关系的不同任务 * <p> * countDownLatch类中只提供了一个构造器：参数count为计数值 * public CountDownLatch(int count) { }; * <p> * 类中有三个方法是最重要的： * await()方法的线程会被挂起，它会等待直到count值为0才继续执行 * public void await() throws InterruptedException { }; * <p> * await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行 * public boolean await(long timeout, TimeUnit unit) throws InterruptedException { }; * <p> * count值减1 * public void countDown() { }; * * @author 20023262 * @date 2020/12/15 * @since 1.0 */
    public class CountDownLatchTest { 
    
        @Test
        public void countDownLatchTest2() { 
            int threadCount = 30;
            CountDownLatch countDownLatch = new CountDownLatch(threadCount);
            for (int i = 0; i < threadCount; i++) { 
                CountDownLatchDemo countDownLatchDemo = new CountDownLatchDemo(countDownLatch);
                ThreadPoolUtil.abortPolicyPool.execute(countDownLatchDemo);
            }
    
            try { 
                // 主线程挂起了，只有等到countDownLatch为0的时候才会唤醒主线程，等待cpu资源执行
                countDownLatch.await();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            System.out.println("子线程执行完成，继续执行主线程");
    
            // 关闭线程池
            ThreadPoolUtil.abortPolicyPool.shutdown();
        }
    
    
        /** * 这种情况适合明确需要开启多少线程 * 比如就需要2个线程并行执行完才能执行下边的方法，且两个并行方法没有依赖关系 */
        @Test
        public void countDownLatchTest() { 
            // 创建门栓
            CountDownLatch countDownLatch = new CountDownLatch(2);
            System.out.println("主线程开始执行...");
    
            /** * 开启第一个线程 */
            ThreadPoolUtil.abortPolicyPool.execute(() -> { 
                try { 
                    Thread.sleep(3000);
                    System.out.println(Thread.currentThread().getName() + "线程执行完成");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    countDownLatch.countDown();
                }
            });
    
            /** * 开启第二个线程 */
            ThreadPoolUtil.abortPolicyPool.execute(() -> { 
                try { 
                    Thread.sleep(2000);
                    System.out.println(Thread.currentThread().getName() + "线程执行完成");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    countDownLatch.countDown();
                }
            });
    
            System.out.println("等待两个线程执行完成");
            // 等待两个线程执行完成
            try { 
                // 主线程挂起了，只有等到countDownLatch为0的时候才会唤醒主线程，等待cpu资源执行
                countDownLatch.await();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            System.out.println("两个线程执行完成，继续执行主线程");
    
            // 关闭线程池
            ThreadPoolUtil.abortPolicyPool.shutdown();
        }
    
    }

**CyclicBarrier**

import java.util.concurrent.CyclicBarrier;
    
    /** * CyclicBarrier测试类 * * @author 20023262 * @date 2020/12/15 * @since 1.0 */
    public class CyclicBarrierDemo implements Runnable { 
    
        private CyclicBarrier cyclicBarrier;
    
        public CyclicBarrierDemo(CyclicBarrier cyclicBarrier) { 
            this.cyclicBarrier = cyclicBarrier;
        }
    
        /** * 在所有线程创建完成之前不会开始执行 * 也就是说每一个线程会等待所有线程全部创建完成之后，统一并发执行 * 大概过程：顺序创建线程 -> 等待所有线程创建完成 -> 并发执行线程 */
        @Override
        public void run() { 
            try { 
                System.out.println(Thread.currentThread().getName() + "开始等待其他线程");
                // 先来的线程挂起，等待其他线程到来
                cyclicBarrier.await();
    
                System.out.println(Thread.currentThread().getName() + "开始执行");
                // 工作线程开始处理，这里用Thread.sleep()来模拟业务处理
                Thread.sleep(1000);
                System.out.println(Thread.currentThread().getName() + "执行完毕");
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

/** * CyclicBarrier测试类 * * @author 20023262 * @date 2020/12/15 * @since 1.0 */
    public class CyclicBarrierTest { 
        @Test
        public void cyclicBarrierTest() { 
            int threadCount = 3;
            CyclicBarrier cyclicBarrier = new CyclicBarrier(threadCount);
            for (int i = 0; i < threadCount; i++) { 
                System.out.println("创建工作线程" + i);
                CyclicBarrierDemo cyclicBarrierDemo = new CyclicBarrierDemo(cyclicBarrier);
                ThreadPoolUtil.abortPolicyPool.execute(cyclicBarrierDemo);
            }
            // 关闭线程池
            ThreadPoolUtil.abortPolicyPool.shutdown();
        }
    }

**Semaphore**

/** * 信号量 * Semaphore 通常我们叫它信号量， 可以用来控制同时访问特定资源的线程数量，通过协调各个线程，以保证合理的使用资源。 * 一般用于限流，控制访问量，比如停车场 * * @author 20023262 * @date 2020/12/15 * @since 1.0 */
    public class SemaphoreDemo implements Runnable { 
    
        private Semaphore semaphore;
    
        public SemaphoreDemo(Semaphore semaphore) { 
            this.semaphore = semaphore;
        }
    
        @Override
        public void run() { 
            try { 
                //获取许可
                semaphore.acquire();
                //执行
                System.out.println("执行线程：" + Thread.currentThread().getName());
                Thread.sleep(1000);
                //释放
                semaphore.release();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

/** * 信号量测试类 * * @author 20023262 * @date 2020/12/15 * @since 1.0 */
    public class SemaphoreTest { 
        /** * 开启10个线程通过信号量控制限定3个线程访问 */
        @Test
        public void semaphoreTest() { 
            int threadCount = 3;
            Semaphore semaphore = new Semaphore(threadCount);
    
            for (int i = 0; i < 10; i++) { 
                System.out.println("创建线程" + i);
                SemaphoreDemo semaphoreDemo = new SemaphoreDemo(semaphore);
                ThreadPoolUtil.abortPolicyPool.execute(semaphoreDemo);
            }
            System.out.println("执行完成");
            // 关闭线程池
            ThreadPoolUtil.abortPolicyPool.shutdown();
        }
    }