多线程 ExecutorService CountDownLatch使用

约定不等于承诺〃 2023-07-15 09:10 21阅读 0赞

一 Java通过Executors提供四种线程池，分别为：   
newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。   
newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。   
newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。   
newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

二、 ExecutorService 的submit（） 与execute（）区别   
1、 submit（）可以接收runnable无返回值和callable有返回值   
      execute（）接收runnable 无返回值

2、submit有返回值，而execute没有

三、关闭线程池方法 shotdown（） showdownNow（）区别

当我们使用shutdownNow方法关闭线程池时，一定要对任务里进行异常捕获。

当我们使用shuwdown方法关闭线程池时，一定要确保任务里不会有永久阻塞等待的逻辑，否则线程池就关闭不了。

最后，一定要记得，shutdownNow和shuwdown调用完，线程池并不是立马就关闭了，要想等待线程池关闭，还需调用awaitTermination方法来阻塞等待。

shutdown() 关闭了提交通道，用submit()是无效（停止接收新任务）原来的任务继续执行  
shutdownNow() 停止接收新任务，原来的任务停止执行

CountDownLatch是JDK提供给我们的多线程间通信的一个工具，用于让主线程知道任务完成的进度。  
如果需要等待所有的线程在执行完后触发一个操作，这个时候你就需要用到CountDownLatch。

//调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行  
public void await() throws InterruptedException \{ \};     
//和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行  
public boolean await(long timeout, TimeUnit unit) throws InterruptedException \{ \};    
//将count值减1  
public void countDown() \{ \};

public void doSome(int num){
         // 创建一个线程池,线程数2
         ExecutorService pool = Executors.newFixedThreadPool(2);
         // CountDownLatch用来监控子线程
         CountDownLatch endSigle = new CountDownLatch(2);
         Callable<Xxaaa> xxxCall = new Xxaaa(a);
         Callable<Xxaaa1> xxxCall1 = new Xxaaa1(a);
         Future<Xxaaa> ocrSearchFuture = pool.submit(xxxCall);//1
         Future<Xxaaa> ocrSearchFuture = pool.submit(xxxCall1);//2
         Xxaaa xxaaa =ocrSearchFuture.get()
         try{
             // 等待两个线程执行结束     
             endSigle.await();
             new Thread(new UserCall(paramMap,.......).start();
         }
         catch (Exception e){
             LOGGER.info(e);
             return null;
         }
         finally{
             if (null != pool && !pool.isShutdown()){
                 pool.shutdown();
             }
         }
     }
     public class Xxaaa implements Callable<Xxaaa>{
         private String a;
         public OcrSearchCall(a){
             this.a = a;
         }
         @SuppressWarnings("finally")
         public xxaaa call(){
             //将count值减1，
             endSigle.countDown();
    
             Xxaaa xxaaa = null;
             return xxaaa;
         }
    
         public class UserCall implements Runnable{
             //基本参数
             private Map<String, Object> paramMap;
             public UserCall(Map<String, Object> paramMap,......){
                 this.paramMap = paramMap;
    ......
             }
    
             @SuppressWarnings("finally")
             public void run(){
                 int a=1;
             }
         }

`实例：`

/* 线程池可以把线程复用起来，减少线程创建销毁的时间和资源消耗，提高了程序任务执行的吞吐率。
        就像线程属于全局使用的资源一样，线程池一般也是全局性，对整个应用进程的线程复用做有效的管理。设计者一般都会把线程池作为类的静态成员或者单例成员，存活于整个进程的生命周期。*/
    
       /* 如果没有设置核心线程数，比如 newCachedThreadPool ，在线程池的线程空闲时间到达 60s 后，线程会关闭，所有线程关闭后线程池也相应关闭回收。
        如果设置了核心线程数，比如 newSingleThreadExecutor 和 newFixedThreadPool ，如果没有主动去关闭，或者设置核心线程的超时时间，核心线程会一直存在不会被关闭，这个线程池就不会被释放回收。
       */
       //newFixedThreadPool线程数设置50，不会被回收
         private   ExecutorService pool = Executors.newFixedThreadPool(2);

@RequestMapping("/testUserInfoCall")
        @ResponseBody
        public String testUserInfoCall(){
            String result="";
            try{
                long startTime=System.currentTimeMillis();
                for (int i = 0; i < 3; i++) {
    
                    result+= getAreaPress();
    
                }
                long endTime=System.currentTimeMillis()-startTime;
                System.out.println("整个方法耗时："+endTime+"ms");
            }catch (Exception e){
                System.out.println(e.getMessage());
            }
            return result;
        }
    
        public String getAreaPress(){
            String result="打印-----=";
            try{
                    long startTime=System.currentTimeMillis();
                    /*　第一：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
                　　 第二：提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
                　　 第三：提高线程的可管理性。*/
                    CountDownLatch flag = new CountDownLatch(2);
                    Callable<Integer> userInfoCall = new UserInfoCall();
                    Future<Integer> userInfoFuture = pool.submit(userInfoCall);
                    int activeCount = ((ThreadPoolExecutor)pool).getActiveCount();
                    long taskCount= ((ThreadPoolExecutor) pool).getTaskCount();
                    long completedTaskCount= ((ThreadPoolExecutor) pool).getCompletedTaskCount();
                    int  getPoolSize=((ThreadPoolExecutor) pool).getPoolSize();
                    int getLargestPoolSize=((ThreadPoolExecutor) pool).getLargestPoolSize();
    
                    System.out.println("活跃线程数:"+activeCount);
                    System.out.println("线程池已经执行的和未执行的任务总数:"+taskCount);
                    System.out.println("线程池当前的线程数量:"+getPoolSize);
                    System.out.println("线程池已完成的任务数量，该值小于等于taskCount:"+completedTaskCount);
                    System.out.println("线程池曾经创建过的最大线程数量。通过这个数据可以知道线程池是否满过，也就是达到了maximumPoolSize:"+getLargestPoolSize);
                    //向线程池中提交任务的submit方法不是阻塞方法，而Future.get方法是一个阻塞方法，
                    // 当submit提交多个任务时，只有所有任务都完成后，才能使用get按照任务的提交顺序得到返回结果，
                    // 所以一般需要使用future.isDone先判断任务是否全部执行完成，完成后再使用future.get得到结果
                    // 。（也可以用get (long timeout, TimeUnit unit)方法可以设置超时时间，防止无限时间的等待）
                    // Integer a = userInfoFuture.get(6000, TimeUnit.MILLISECONDS);
                    //userInfoFuture.get()等待线程返回结果才能继续向下执行-》如果不使用则直接不等线程执行继续往下
                    //userInfoFuture.get();
                    long endTime=System.currentTimeMillis()-startTime;
                    System.out.println("耗时："+endTime+"ms");
    
            }catch (Exception e){
                System.out.println(e.getMessage());
            }
            return result;
        }

注意此处：Future模式是多线程设计常用的一种设计模式。Future模式可以理解成：我有一个任务，提交给了Future，Future替我完成这个任务。期间我自己可以去做任何想做的事情。一段时间之后，我就便可以从Future那儿取出结果。

Future<Integer> userInfoFuture = pool.submit(userInfoCall);

userInfoFuture.get()

![20200317155432837.png][]

final CountDownLatch endSign = new CountDownLatch(3);

endSign .await() 当为0时候才可以继续执行下去

public static void main(String[] args) {
            try{
                List<Future<Integer>> list=null;
                long startTime=System.currentTimeMillis();
                list= getAreaPress111();
                for (int i = 0; i < list.size(); i++) {
                    Integer flag= list.get(i).get();
                    System.out.println("====="+i+"========"+flag);
                }
                long endTime=System.currentTimeMillis()-startTime;
                System.out.println("整个方法耗时："+endTime+"ms");
            }catch (Exception e){
                System.out.println("异常："+e);
            }
        }
    
        private static   List<Future<Integer>> getAreaPress111(){
            List<Future<Integer>> futures = new ArrayList<Future<Integer>>();
            ExecutorService pool = Executors.newFixedThreadPool(3);
            final CountDownLatch endSign = new CountDownLatch(3);
            try {
                for (int i = 0; i < 3; i++) {
                    Callable<Integer> userInfoCall = new UserInfoCall(endSign);
                    Future<Integer> future = pool.submit(userInfoCall);
                 /*   Future<Integer> future = pool.submit(new Callable<Integer>() {
                        @Override
                        public Integer call() throws Exception {
                            Thread.sleep(5000);
                            endSign.countDown();
                            System.out.println("sss"+endSign.getCount());
                            return 1;
                        }
                    });*/
                    futures.add(future);
                }
                //等待三个线程执行结束 才能继续向下执行
                endSign.await();
                pool.shutdown();
            } catch (InterruptedException e) {
                System.out.println(e);
            }
            return futures;
        }

[20200317155432837.png]: /images/20230528/60c78c529c994df99a248b044676a50c.png