三.多线程JUC篇-3.17 ExecutorService API详细讲解

小灰灰 2022-12-31 04:26 76阅读 0赞

## 1.shutdown ##

public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException { 
            isShutDown();
        }
    
        private static void isShutDown() { 
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            executorService.execute(() -> { 
                try { 
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            });
    
            System.out.println(executorService.isShutdown());
            executorService.shutdown();
            System.out.println(executorService.isShutdown());
            executorService.execute(() -> System.out.println("I will be executed after shutDown ???"));
        }
    }
    
    //输出：
    Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.wangwenjun.concurrent.executors.ExecutorServiceExample1$$Lambda$2/558638686@448139f0 rejected from java.util.concurrent.ThreadPoolExecutor@7cca494b[Shutting down, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]
    	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
    	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
    	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
    	at java.util.concurrent.Executors$DelegatedExecutorService.execute(Executors.java:668)
    	at com.wangwenjun.concurrent.executors.ExecutorServiceExample1.isShutDown(ExecutorServiceExample1.java:28)
    	at com.wangwenjun.concurrent.executors.ExecutorServiceExample1.main(ExecutorServiceExample1.java:9)
    false
    true

## 2.terminate ##

public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException { 
           isTerminated();
        }
    
        private static void isTerminated() { 
            ExecutorService executorService = Executors.newFixedThreadPool(1);
            executorService.execute(() -> { 
                try { 
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            });
            executorService.shutdown();
            System.out.println(executorService.isShutdown());
            System.out.println(executorService.isTerminated());
            System.out.println(((ThreadPoolExecutor)executorService).isTerminating());
        }
    }
    //输出：在这里插入代码片 
    true
    false
    true

## 3.捕获异常 ##

### 3.1 setUncaughtExceptionHandler ###

public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException { 
            executeRunnableError();
        }
    
        private static void executeRunnableError() throws InterruptedException { 
            ExecutorService executorService = Executors.newFixedThreadPool(2, new MyThreadFactory());
            IntStream.range(0, 10).boxed().forEach(i -> executorService.execute(() -> System.out.println(1 / 0)));
            executorService.shutdown();
            executorService.awaitTermination(10, TimeUnit.MINUTES);
            System.out.println("=====================================");
        }
    
        private static class MyThreadFactory implements ThreadFactory { 
            private final static AtomicInteger SEQ = new AtomicInteger();
    
            @Override
            public Thread newThread(Runnable r) { 
                Thread t = new Thread(r);
                t.setName("My-Thread-" + SEQ.getAndIncrement());
                t.setUncaughtExceptionHandler((i, cause) -> { 
                    System.out.println("The thread " + i.getName() + " execute failed.");
                   // cause.printStackTrace();
                    System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
                });
                return t;
            }
    
        }
    }
    
    //输出
    =====================================
    The thread My-Thread-1 execute failed.
    The thread My-Thread-0 execute failed.
    @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
    @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@

### 3.2 线程内部分步骤 ###

public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException { 
            executeRunnableTask();
        }
    
        /* * <pre> * |--->操作1 * |--->操作2 * send request-->store db-->10->|--->操作3-->给出不同的反馈 * |--->操作4 * |--->操作5 * </pre> */
        private static void executeRunnableTask() throws InterruptedException { 
            ExecutorService executorService = Executors.newFixedThreadPool(2);
            IntStream.range(0, 2).boxed().forEach(i -> executorService.execute(
                    new MyTask(i) { 
    
                        @Override
                        protected void error(Throwable cause) { 
                            System.out.println("The no:" + i + " failed , update status to ERROR.");
                        }
    
                        @Override
                        protected void done() { 
                            System.out.println("The no:" + i + " successfully , update status to DONE.");
                        }
    
                        @Override
                        protected void doInit() { 
                            //do nothing
                        }
    
                        @Override
                        protected void doExecute() { 
                            if(i % 3 == 0) { 
                                int tmp = 1 / 0;
                            }
                        }
                    }
            ));
            executorService.shutdown();
            executorService.awaitTermination(10, TimeUnit.MINUTES);
            System.out.println("=====================================");
        }
    
        private static class MyThreadFactory implements ThreadFactory { 
            private final static AtomicInteger SEQ = new AtomicInteger();
    
            @Override
            public Thread newThread(Runnable r) { 
                Thread t = new Thread(r);
                t.setName("My-Thread-" + SEQ.getAndIncrement());
                t.setUncaughtExceptionHandler((i, cause) -> { 
                    System.out.println("The thread " + i.getName() + " execute failed.");
                    cause.printStackTrace();
                    System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
                });
                return t;
            }
    
        }
    
        private abstract static class MyTask implements Runnable { 
            protected final int no;
    
            public MyTask(int no) { 
                this.no = no;
            }
    
            @Override
            public void run() { 
                try { 
                    this.doInit();
                    this.doExecute();
                    this.done();
                } catch (Throwable cause) { 
                    this.error(cause);
                }
            }
    
            protected abstract void error(Throwable cause);
    
            protected abstract void done();
    
            protected abstract void doExecute();
    
            protected abstract void doInit();
        }
    }
    
    //输出：
    The no:1 successfully , update status to DONE.
    The no:0 failed , update status to ERROR.
    =====================================

## 4.拒绝策略 ##

### 4.1 线程池大小足够 ###

public class ExecutorServiceExample1 { 
        public static void main(String[] args) { 
    
            ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 3, 10, TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<Runnable>(1));
            for (int i = 0; i < 4; i++) { 
                MyTask myTask = new MyTask(i);
                executor.execute(myTask);
                System.out.println("线程池中线程数目：" + executor.getPoolSize() + "，队列中等待执行的任务数目：" +
                        executor.getQueue().size() + "，已执行完的任务数目：" + executor.getCompletedTaskCount());
            }
            executor.shutdown();
        }
    }
    
    class MyTask implements Runnable { 
        private int taskNum;
    
        public MyTask(int num) { 
            this.taskNum = num;
        }
    
        @Override
        public void run() { 
            System.out.println("线程名称：" + Thread.currentThread().getName() + "，正在执行task " + taskNum);
            try { 
                Thread.currentThread().sleep(4000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            System.out.println("task " + taskNum + "执行完毕");
    
        }
    }
    
    //输出：
    线程池中线程数目：1，队列中等待执行的任务数目：0，已执行完的任务数目：0
    线程名称：pool-1-thread-1，正在执行task 0
    线程池中线程数目：2，队列中等待执行的任务数目：0，已执行完的任务数目：0
    线程池中线程数目：2，队列中等待执行的任务数目：1，已执行完的任务数目：0
    线程池中线程数目：3，队列中等待执行的任务数目：1，已执行完的任务数目：0
    线程名称：pool-1-thread-2，正在执行task 1
    线程名称：pool-1-thread-3，正在执行task 3
    task 0执行完毕
    线程名称：pool-1-thread-1，正在执行task 2
    task 1执行完毕
    task 3执行完毕
    task 2执行完毕
    
    Process finished with exit code 0

### 4.2 AbortPolicy ###

//将任务数调整为5
    //超过的任务（在这里有一个任务）会被拒绝，并抛出任何异常
    线程池中线程数目：1，队列中等待执行的任务数目：0，已执行完的任务数目：0
    线程池中线程数目：2，队列中等待执行的任务数目：0，已执行完的任务数目：0
    线程池中线程数目：2，队列中等待执行的任务数目：1，已执行完的任务数目：0
    线程池中线程数目：3，队列中等待执行的任务数目：1，已执行完的任务数目：0
    线程名称：pool-1-thread-1，正在执行task 0
    线程名称：pool-1-thread-2，正在执行task 1
    线程名称：pool-1-thread-3，正在执行task 3
    Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.wangwenjun.concurrent.executors.MyTask@135fbaa4 rejected from java.util.concurrent.ThreadPoolExecutor@45ee12a7[Running, pool size = 3, active threads = 3, queued tasks = 1, completed tasks = 0]
    	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
    	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
    	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
    	at com.wangwenjun.concurrent.executors.ExecutorServiceExample1.main(ExecutorServiceExample1.java:14)
    task 0执行完毕
    线程名称：pool-1-thread-1，正在执行task 2
    task 3执行完毕
    task 1执行完毕
    task 2执行完毕

### 4.3 DiscardPolicy ###

//超过的任务（在这里有一个任务）会被拒绝，但不会抛出任何异常
    线程池中线程数目：1，队列中等待执行的任务数目：0，已执行完的任务数目：0
    线程池中线程数目：2，队列中等待执行的任务数目：0，已执行完的任务数目：0
    线程池中线程数目：2，队列中等待执行的任务数目：1，已执行完的任务数目：0
    线程池中线程数目：3，队列中等待执行的任务数目：1，已执行完的任务数目：0
    线程池中线程数目：3，队列中等待执行的任务数目：1，已执行完的任务数目：0
    线程名称：pool-1-thread-2，正在执行task 1
    线程名称：pool-1-thread-1，正在执行task 0
    线程名称：pool-1-thread-3，正在执行task 3
    task 1执行完毕
    task 0执行完毕
    task 3执行完毕
    线程名称：pool-1-thread-2，正在执行task 2
    task 2执行完毕

### 4.4 DiscardOldestPolicy ###

当任务被拒绝添加时，会抛弃任务队列中最旧的任务也就是最先加入队列的，再把这个新任务添加进去。

### 4.4 CallerRunsPolicy ###

//由调用线程执行，而不是ThreadPoolExecute
    线程池中线程数目：1，队列中等待执行的任务数目：0，已执行完的任务数目：0
    线程池中线程数目：2，队列中等待执行的任务数目：0，已执行完的任务数目：0
    线程池中线程数目：2，队列中等待执行的任务数目：1，已执行完的任务数目：0
    线程池中线程数目：3，队列中等待执行的任务数目：1，已执行完的任务数目：0
    线程名称：main，正在执行task 4
    线程名称：pool-1-thread-1，正在执行task 0
    线程名称：pool-1-thread-2，正在执行task 1
    线程名称：pool-1-thread-3，正在执行task 3
    task 4执行完毕
    task 1执行完毕
    task 0执行完毕
    线程名称：pool-1-thread-2，正在执行task 2
    task 3执行完毕
    线程池中线程数目：3，队列中等待执行的任务数目：1，已执行完的任务数目：0
    task 2执行完毕

### 5. ActiveCount ###

public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException { 
            test();
        }
    
        private static void test() throws InterruptedException { 
            ThreadPoolExecutor executorService = (ThreadPoolExecutor)Executors.newFixedThreadPool(5);
            System.out.println(executorService.getActiveCount());
    
            executorService.execute(() -> { 
                try { 
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            });
    
            TimeUnit.MILLISECONDS.sleep(20);
            System.out.println(executorService.getActiveCount());
    
        }
    }
    
    //输出
    0
    1

### 5. AllowCoreThreadTimeOut ###

//如果不设置allowCoreThread，pool会一直存在，因为core threads一直为5
    //如果设置了allowCoreThread，等到3秒会，pool会停止
    private static void testAllowCoreThreadTimeOut() { 
            ThreadPoolExecutor executorService = (ThreadPoolExecutor)Executors.newFixedThreadPool(5);
            executorService.setKeepAliveTime(3, TimeUnit.SECONDS); 
            executorService.allowCoreThreadTimeOut(true); //为true时keepAliveTime的时间必须大于0
            IntStream.range(0, 5).boxed().forEach(i -> { 
                executorService.execute(() -> { 
                    try { 
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
            });
        }

## 6.remove ##

public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException { 
            testRemove();
        }
    
        private static void testRemove() throws InterruptedException { 
            ThreadPoolExecutor executorService = (ThreadPoolExecutor)Executors.newFixedThreadPool(2);
            executorService.setKeepAliveTime(10, TimeUnit.SECONDS);
            //为true时keepAliveTime的时间必须大于1
            executorService.allowCoreThreadTimeOut(true);
            IntStream.range(0, 2).boxed().forEach(i -> { 
                executorService.execute(() -> { 
                    try { 
                        TimeUnit.SECONDS.sleep(5);
                        System.out.println("================ I am finished.");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
            });
    
            TimeUnit.MILLISECONDS.sleep(20);
            Runnable r = () -> { 
                System.out.println("I will never be executed.");
            };
            executorService.execute(r);
            TimeUnit.MILLISECONDS.sleep(20);
            //将queue中的任务删除
            executorService.remove(r);
        }
    }
    
    //输出：
    ================ I am finished.
    ================ I am finished.

## 7.beforeExecute和afterExecute ##

public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException { 
            testThreadPoolAdvice();
        }
    
    
    
        //beforeExecute和afterExecute类似于切面
        private static void testThreadPoolAdvice() { 
            ThreadPoolExecutor executorService = new MyThreadPoolExecutor(1, 2, 30, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(1), r -> { 
                Thread t = new Thread(r);
                return t;
            }, new ThreadPoolExecutor.AbortPolicy());
    
            executorService.execute(new MyRunnable(1) { 
                @Override
                public void run() { 
                    System.out.println("=================" + 1/0);
                }
            });
        }
    
        private abstract static class MyRunnable implements Runnable { 
            private final int no;
    
            public MyRunnable(int no) { 
                this.no = no;
            }
    
            protected int getData() { 
                return this.no;
            }
        }
    
        private static class MyThreadPoolExecutor extends ThreadPoolExecutor { 
    
            public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                        BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { 
                super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
            }
    
            @Override
            protected void beforeExecute(Thread t, Runnable r) { 
                System.out.println("init the " + ((MyRunnable)r).getData());
            }
    
            @Override
            protected void afterExecute(Runnable r, Throwable t) { 
                if(null == t) { 
                    System.out.println("successful " + ((MyRunnable)r).getData());
                } else { 
                    System.out.println("failed " + t.getMessage());
                }
            }
        }
    }
    
    //输出
    init the 1
    failed / by zero
    Exception in thread "Thread-0" java.lang.ArithmeticException: / by zero
    	at com.wangwenjun.concurrent.executors.ExecutorServiceExample1$1.run(ExecutorServiceExample1.java:27)
    	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    	at java.lang.Thread.run(Thread.java:748)

## 8.invoke和submit ##

### 8.1 InvokeAny ###

//InvokeAny会阻塞主线程，直到某一线程执行完毕
    //InvokeAny有一个线程执行完毕，其它线程会取消执行，不在执行
    //InvokeAny有一个线程执行完毕，如果其它线程同时执行完毕，就不存在取消一说了
    public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException, ExecutionException { 
            testInvokeAny();
        }
    
    
    
        private static void testInvokeAny() throws InterruptedException, ExecutionException { 
            ExecutorService executorService = Executors.newFixedThreadPool(10);
    
            List<Callable<Integer>> callableList = IntStream.range(0, 5).boxed().map(
                    i -> (Callable<Integer>) () -> { 
                        System.out.println(Thread.currentThread().getName() + " :" + i + " start");
                        TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(20));
                        System.out.println(Thread.currentThread().getName() + " :" + i + " end");
                        return i;
                    }
            ).collect(Collectors.toList());
            //这是一个同步方法，执行完成后才会执行后面的任务
            Integer value = executorService.invokeAny(callableList);
            System.out.println("===================finished=====================");
            System.out.println(value);
        }
    }
    
    //输出：
    pool-1-thread-1 :0 start
    pool-1-thread-3 :2 start
    pool-1-thread-2 :1 start
    pool-1-thread-5 :4 start
    pool-1-thread-4 :3 start
    pool-1-thread-2 :1 end
    ===================finished=====================
    1

### 8.2 InvokeAny timeout ###

public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException { 
            testInvokeAnyTimeOut();
        }
    
    
    
        private static void testInvokeAnyTimeOut() throws InterruptedException, ExecutionException, TimeoutException { 
            ExecutorService executorService = Executors.newFixedThreadPool(10);
    
            List<Callable<Integer>> callableList = IntStream.range(0, 5).boxed().map(
                    i -> (Callable<Integer>) () -> { 
                        System.out.println(Thread.currentThread().getName() + " :" + i + " start");
                        TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(100));
                        System.out.println(Thread.currentThread().getName() + " :" + i + " end");
                        return i;
                    }
            ).collect(Collectors.toList());
            //这是一个同步方法，执行完成后才会执行后面的任务
            Integer value = executorService.invokeAny(callableList, 3, TimeUnit.SECONDS);
            System.out.println("===================finished=====================");
            System.out.println(value);
        }
    }
    
    //输出
    pool-1-thread-1 :0 start
    pool-1-thread-4 :3 start
    pool-1-thread-3 :2 start
    pool-1-thread-2 :1 start
    pool-1-thread-5 :4 start
    Exception in thread "main" java.util.concurrent.TimeoutException
    	at java.util.concurrent.AbstractExecutorService.doInvokeAny(AbstractExecutorService.java:184)
    	at java.util.concurrent.AbstractExecutorService.invokeAny(AbstractExecutorService.java:225)
    	at com.wangwenjun.concurrent.executors.ExecutorServiceExample1.testInvokeAnyTimeOut(ExecutorServiceExample1.java:30)
    	at com.wangwenjun.concurrent.executors.ExecutorServiceExample1.main(ExecutorServiceExample1.java:13)

### 8.3 InvokeAll ###

//InvokeAll同样是个阻塞方法
    //这里发现一个问题，就是callabe执行完后，才进行foreach，这里在jdk1.9已解决或者rxjava框架
    public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException { 
            testInvokeAll();
        }
    
    
    
        private static void testInvokeAll() throws InterruptedException { 
            ExecutorService executorService = Executors.newFixedThreadPool(10);
    
            //这是一个同步方法，执行完成后才会执行后面的任务
            executorService.invokeAll(
                    IntStream.range(0, 5).boxed().map(
                            i -> (Callable<Integer>) () -> { 
                                TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(30));
                                System.out.println(Thread.currentThread().getName() + " :" + i);
                                return i;
                            }
                    ).collect(Collectors.toList())
            ).stream().map(future -> { 
                try { 
                    return future.get();
                } catch (Exception e) { 
                    throw new RuntimeException();
                }
            }).forEach(System.out::println);
    
            System.out.println("===================finished=====================");
        }
    }
    
    //输出：
    pool-1-thread-1 :0
    pool-1-thread-5 :4
    pool-1-thread-3 :2
    pool-1-thread-4 :3
    pool-1-thread-2 :1
    0
    1
    2
    3
    4
    ===================finished=====================

### 8.5 submit ###

public class ExecutorServiceExample1 { 
        public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException { 
            testSubmitRunnable();
            testSubmitRunnableWithResult();
        }
    
    
    
        /** * {@link ExecutorService#submit(Runnable)} */
        private static void testSubmitRunnable() throws InterruptedException, ExecutionException { 
            ExecutorService executorService = Executors.newFixedThreadPool(10);
            Future<?> future = executorService.submit(() -> { 
                try { 
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println(Thread.currentThread().getName());
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            });
            //get()会产生阻塞，等待前面的任务执行完
            Object NULL = future.get();
            System.out.println("R:" + NULL);
        }
    
        /** * {@link ExecutorService#submit(Runnable, Object)} */
        private static void testSubmitRunnableWithResult() throws InterruptedException, ExecutionException { 
            ExecutorService executorService = Executors.newFixedThreadPool(10);
            String result = "DONE";
            Future<String> future = executorService.submit(() -> { 
                try { 
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println(Thread.currentThread().getName());
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }, result);
            result = future.get();
            System.out.println("R:" + result);
        }
    }
    
    //输出
    pool-1-thread-1
    R:null
    
    pool-2-thread-1
    R:DONE