ToB企服应用市场:ToB评测及商务社交产业平台

标题: Juc并发编程12——2万字深入源码：线程池这篇真的讲解的透透的了 [打印本页]
作者: 涛声依旧在    时间: 2022-6-23 10:29
标题: Juc并发编程12——2万字深入源码：线程池这篇真的讲解的透透的了
本文将介绍常见的线程池的使用方法，介绍线程池的参数、拒绝策略、返回参数获取以及定时调度。

2万字深入源码：线程池这篇真的讲解的透透的了



1.线程池介绍

利用多线程我们可以合理的利用cpu资源，更加高效的完成工作，不过如果我们频繁的创建、销毁线程，也会对系统的资源造成消耗。
因此，我们也可以利用池化技术，就像数据库的连接池一样，使用线程池来管理多个线程，不对这些线程进行销毁，然后反复的使用这些线程。在Tomcat服务器中，可能同一时间服务器会收到大量的请求，频繁的创建、销毁线程肯定不可行，因此就使用了线程池技术。
线程池一般具有容量限制，如果所有的线程都处于工作状态，在接受到新的多线程请求时，将会进入阻塞状态，直到有可用的空闲线程，实际上就是使用阻塞队列处理的。

2.线程池的使用

2.1 构造方法参数详解

我们可以使用ThreadPoolExecutor对象来创建线程池，先看看它的构造方法。
  1. public ThreadPoolExecutor(int corePoolSize,
  2.                               int maximumPoolSize,
  3.                               long keepAliveTime,
  4.                               TimeUnit unit,
  5.                               BlockingQueue<Runnable> workQueue,
  6.                               ThreadFactory threadFactory,
  7.                               RejectedExecutionHandler handler) {
  8.         if (corePoolSize < 0 ||
  9.             maximumPoolSize <= 0 ||
  10.             maximumPoolSize < corePoolSize ||
  11.             keepAliveTime < 0)
  12.             throw new IllegalArgumentException();
  13.         if (workQueue == null || threadFactory == null || handler == null)
  14.             throw new NullPointerException();
  15.         this.acc = System.getSecurityManager() == null ?
  16.                 null :
  17.                 AccessController.getContext();
  18.         this.corePoolSize = corePoolSize;
  19.         this.maximumPoolSize = maximumPoolSize;
  20.         this.workQueue = workQueue;
  21.         this.keepAliveTime = unit.toNanos(keepAliveTime);
  22.         this.threadFactory = threadFactory;
  23.         this.handler = handler;
  24.     }
复制代码
其输出如下。

我们看看i的输出顺序，似乎有点奇怪，我们来分析下。首先i为0，1的线程会先执行，其次i为3，4的线程进入等待队列，最后i为5，6的线程直接对线程池扩容执行，此时线程池满，i为3，4的线程等待有空闲线程时才会执行。所以就是0145先执行，23后执行。至于0，1，4，5或2，3两组内部的执行顺序这里留个坑。
我们还会发现，上面的程序怎么不退出呢？这是因为我们没有销毁线程池。里面的核心线程会一直处于等待状态。来关闭下。
  1. public ThreadPoolExecutor(int corePoolSize,
  2.                               int maximumPoolSize,
  3.                               long keepAliveTime,
  4.                               TimeUnit unit,
  5.                               BlockingQueue<Runnable> workQueue) {
  6.         this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
  7.              Executors.defaultThreadFactory(), defaultHandler);
  8.     }
复制代码
上面我们等待队列使用的是容量为2的ArrayBlockingQueue,我们把它换成SynchronousQueue看看会发生什么？
  1. public static void main(String[] args) throws InterruptedException {
  2.         //corePoolSize:2 ;maximumPoolSize:4; keepAliveTime:3s;workQueueSize:2
  3.         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2));
  4.         for (int i = 0; i < 6; i++) {
  5.             int finalI = i;
  6.             executor.execute(() -> {
  7.                 System.out.println("prepare to execute thread :" + finalI);
  8.                 try {
  9.                     TimeUnit.SECONDS.sleep(1);
  10.                 } catch (InterruptedException exception) {
  11.                     exception.printStackTrace();
  12.                 }
  13.                 System.out.println("finish to execute thread :" + finalI);
  14.             });
  15.         }
  17.         TimeUnit.SECONDS.sleep(1);
  18.         System.out.println("pool size:" + executor.getPoolSize()); //查看当前线程池中线程数,此时应该最多有4个线程，因为线程池最大容量是4
  19.         TimeUnit.SECONDS.sleep(5); //keepAliveTime已过
  20.         System.out.println("pool size:" + executor.getPoolSize());
  21.     }
复制代码
输出如下。
在线程0，1，2，3执行完后，报出来的异常信息是java.util.concurrent.RejectedExecutionException，它执行了拒绝策略。我们来分析下，首先1，2执行没问题，3，4首先想要进入等待队列，不过SynchronousQueue根本没有容量，相当于等待队列满了，因此开始进行创建新的线程池，等5，6来时，线程池已经达到最大的线程数，因此执行了拒绝策略。这里的默认拒绝策略使用的是抛出异常。我们也可以修改。并且上面的程序不会退出，原因是啥笔者还没有弄明白，如果有大佬知道欢迎在评论区不吝赐教哟。
2.3 线程池的拒绝策略

线程池的拒绝策略有以下几个。

先来测试下CallerRunsPolicy
  1.   public static void main(String[] args) throws InterruptedException {
  2.         //
  3.         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2));
  4.         for (int i = 0; i < 6; i++) {
  5.             int finalI = i;
  6.             executor.execute(() -> {
  7.                 System.out.println("prepare to execute thread " + finalI);
  8.                 try {
  9.                     TimeUnit.SECONDS.sleep(1);
  10.                 } catch (InterruptedException exception) {
  11.                     exception.printStackTrace();
  12.                 }
  13.                 System.out.println("finish to execute thread " + finalI);
  14.             });
  15.         }
  17.         TimeUnit.SECONDS.sleep(1);
  18.         System.out.println("pool size:" + executor.getPoolSize());
  19.         TimeUnit.SECONDS.sleep(5);
  20.         System.out.println("pool size:" + executor.getPoolSize());
  22.         executor.shutdownNow(); //关闭线程池，取消等待队列中的任务，试图终止正在执行的任务，不再接收新的任务
  23.         // executor.shutdown(); 执行完等待队列中的任务再关闭
  24.     }
复制代码
其执行结果如下。

可以看到i为4的线程确实是由主线程执行的哟。等它在主线程执行完的时候，正好线程池也有空闲线程了，因此就又可以让线程池的线程执行任务了。
接下来演示下DiscardOldestPolicy。
  1. public static void main(String[] args) throws InterruptedException {
  2.         //
  3.         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new SynchronousQueue<>());
  4.         for (int i = 0; i < 6; i++) {
  5.             int finalI = i;
  6.             executor.execute(() -> {
  7.                 System.out.println("prepare to execute thread " + finalI);
  8.                 try {
  9.                     TimeUnit.SECONDS.sleep(1);
  10.                 } catch (InterruptedException exception) {
  11.                     exception.printStackTrace();
  12.                 }
  13.                 System.out.println("finish to execute thread " + finalI);
  14.             });
  15.         }
  17.         TimeUnit.SECONDS.sleep(1);
  18.         System.out.println("pool size:" + executor.getPoolSize());
  19.         TimeUnit.SECONDS.sleep(5);
  20.         System.out.println("pool size:" + executor.getPoolSize());
  22.         executor.shutdownNow();
  23.     }
复制代码
其输出如下。

Thread2没有执行，这是因为它是最早进入等待队列的，在最后一个任务提交时，使用拒绝策略DiscardOldestPolicy将其取消了。
如果使用SynchronousQueue做等待队列，DiscardOldestPolicy做拒绝策略，会出现什么情况呢？
  1. public static void main(String[] args) throws InterruptedException {
  2.         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new SynchronousQueue<>(), new ThreadPoolExecutor.CallerRunsPolicy());
  3.         for (int i = 0; i < 6; i++) {
  4.             int finalI = i;
  5.             executor.execute(() -> {
  6.                 System.out.println("prepare to execute thread i:"+ finalI  + " ,name:"+ Thread.currentThread().getName());
  7.                 try {
  8.                     TimeUnit.SECONDS.sleep(1);
  9.                 } catch (InterruptedException exception) {
  10.                     exception.printStackTrace();
  11.                 }
  12.                 System.out.println("finish to execute thread " + finalI  + " ,name:"+Thread.currentThread().getName());
  13.             });
  14.         }
  16.         TimeUnit.SECONDS.sleep(1);
  17.         System.out.println("pool size:" + executor.getPoolSize());
  18.         TimeUnit.SECONDS.sleep(5);
  19.         System.out.println("pool size:" + executor.getPoolSize());
  21.         executor.shutdownNow();
  22.     }
复制代码
输出结果如下。
居然直接爆栈了。为什么呢？我们得看看它的源码了。
  1. public static void main(String[] args) throws InterruptedException {
  2.         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2), new ThreadPoolExecutor.DiscardOldestPolicy());
  3.         for (int i = 0; i < 7; i++) {
  4.             int finalI = i;
  5.             executor.execute(() -> {
  6.                 System.out.println("prepare to execute thread i:"+ finalI );
  7.                 try {
  8.                     TimeUnit.SECONDS.sleep(1);
  9.                 } catch (InterruptedException exception) {
  10.                     exception.printStackTrace();
  11.                 }
  12.                 System.out.println("finish to execute thread " + finalI );
  13.             });
  14.         }
  16.         TimeUnit.SECONDS.sleep(1);
  17.         System.out.println("pool size:" + executor.getPoolSize());
  18.         TimeUnit.SECONDS.sleep(5);
  19.         System.out.println("pool size:" + executor.getPoolSize());
  21.         executor.shutdownNow();
  22.     }
复制代码
原因找到了，原来e.getQueue().poll()对于SynchronousQueue没有任何的意义，它会陷入死循环，一遍遍的执行execute，再执行拒绝策略，最后导致栈溢出。
最后演示下DiscardPolicy。
  1. public static void main(String[] args) throws InterruptedException {
  2.         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new SynchronousQueue<>(), new ThreadPoolExecutor.DiscardOldestPolicy());
  3.         for (int i = 0; i < 6; i++) {
  4.             int finalI = i;
  5.             executor.execute(() -> {
  6.                 System.out.println("prepare to execute thread i:"+ finalI );
  7.                 try {
  8.                     TimeUnit.SECONDS.sleep(1);
  9.                 } catch (InterruptedException exception) {
  10.                     exception.printStackTrace();
  11.                 }
  12.                 System.out.println("finish to execute thread " + finalI );
  13.             });
  14.         }
  16.         TimeUnit.SECONDS.sleep(1);
  17.         System.out.println("pool size:" + executor.getPoolSize());
  18.         TimeUnit.SECONDS.sleep(5);
  19.         System.out.println("pool size:" + executor.getPoolSize());
  21.         executor.shutdownNow();
  22.     }
复制代码
结果如下。真的太简单了，不讲解了。

除了官方提供的拒绝策略外，我们也可以使用自定义的拒绝策略哟。
  1.     public static class DiscardOldestPolicy implements RejectedExecutionHandler {
  2.       
  3.         public DiscardOldestPolicy() { }
  5.         public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  6.             if (!e.isShutdown()) { //如果线程池没有关闭
  7.                 e.getQueue().poll(); //将等待队列队顶中元素出队
  8.                 e.execute(r); //再调用线程池的execute方法
  9.             }
  10.         }
  11.     }
复制代码
运行的结果如下。

2.4 线程创建工厂

我们之前也提到过，我们可以利用线程创建工厂干涉线程的创建过程。
  1. public static void main(String[] args) throws InterruptedException {
  2.         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new SynchronousQueue<>(), new ThreadPoolExecutor.DiscardPolicy());
  3.         for (int i = 0; i < 6; i++) {
  4.             int finalI = i;
  5.             executor.execute(() -> {
  6.                 System.out.println("prepare to execute thread i:"+ finalI );
  7.                 try {
  8.                     TimeUnit.SECONDS.sleep(1);
  9.                 } catch (InterruptedException exception) {
  10.                     exception.printStackTrace();
  11.                 }
  12.                 System.out.println("finish to execute thread " + finalI );
  13.             });
  14.         }
  16.         TimeUnit.SECONDS.sleep(1);
  17.         System.out.println("pool size:" + executor.getPoolSize());
  18.         TimeUnit.SECONDS.sleep(5);
  19.         System.out.println("pool size:" + executor.getPoolSize());
  21.         executor.shutdownNow();
  22.     }
复制代码
输出结果如下。

2.5 线程异常

如果线程池中的线程遇到异常会出现什么情况呢，会不会被销毁呢？
  1. public static void main(String[] args) throws InterruptedException {
  2.         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new SynchronousQueue<>(), (r, e) -> {
  3.             System.out.println("full,full,full!");
  4.             r.run(); //直接在当前的(调用提交方法所在)线程运行
  5.         });
  6.         for (int i = 0; i < 6; i++) {
  7.             int finalI = i;
  8.             executor.execute(() -> {
  9.                 System.out.println("prepare to execute thread i:" + finalI + " ,name:" + Thread.currentThread().getName());
  10.                 try {
  11.                     TimeUnit.SECONDS.sleep(1);
  12.                 } catch (InterruptedException exception) {
  13.                     exception.printStackTrace();
  14.                 }
  15.                 System.out.println("finish to execute thread " + finalI + " ,name:" + Thread.currentThread().getName());
  16.             });
  17.         }
  19.         TimeUnit.SECONDS.sleep(1);
  20.         System.out.println("pool size:" + executor.getPoolSize());
  21.         TimeUnit.SECONDS.sleep(5);
  22.         System.out.println("pool size:" + executor.getPoolSize());
  24.         executor.shutdownNow();
  25.     }
复制代码
输出如下，两次输出的线程名字不同。看来线程池中抛异常的确实会被销毁哟。

3.使用Executors创建线程池

3.1 newFixedThreadPool

除了使用new一个ThreadPoolExecutor的方法，我们还可以使用Executors创建一个线程池。
  1.   public static void main(String[] args) throws InterruptedException {
  2.         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2), new ThreadFactory() {
  3.             int count = 0;
  4.             @Override
  5.             public Thread newThread(Runnable r) { //改变了线程的命名
  6.                 return new Thread(r,"thread" + count++);
  7.             }
  8.         });
  9.         for (int i = 0; i < 6; i++) {
  10.             int finalI = i;
  11.             executor.execute(() -> {
  12.                 System.out.println("prepare to execute thread i:" + finalI + " ,name:" + Thread.currentThread().getName());
  13.                 try {
  14.                     TimeUnit.SECONDS.sleep(1);
  15.                 } catch (InterruptedException exception) {
  16.                     exception.printStackTrace();
  17.                 }
  18.                 System.out.println("finish to execute thread " + finalI + " ,name:" + Thread.currentThread().getName());
  19.             });
  20.         }
  22.         TimeUnit.SECONDS.sleep(1);
  23.         System.out.println("pool size:" + executor.getPoolSize());
  24.         TimeUnit.SECONDS.sleep(5);
  25.         System.out.println("pool size:" + executor.getPoolSize());
  27.         executor.shutdownNow();
  28.     }
复制代码
对比下之前的创建方式。即使是参数最少的构造方法，Executors创建也比new一个ThreadPoolExecutor更加简洁。
  1. public static void main(String[] args) throws InterruptedException {
  2.         ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new LinkedBlockingDeque<>());
  3.         executor.execute(() -> {
  4.             System.out.println(Thread.currentThread().getName());
  5.             throw new RuntimeException("error");
  6.         });
  7.         TimeUnit.SECONDS.sleep(1);
  8.         executor.execute(() -> {
  9.             System.out.println(Thread.currentThread().getName());
  10.         });
  11.     }
  12.     executor.shutdown();
复制代码
再看看源码。就是说包装是yyds。
  1.   public static void main(String[] args) throws InterruptedException{
  2.         // 创建容量固定为10的线程池
  3.         ExecutorService service = Executors.newFixedThreadPool(10); //ThreadPoolExecutor extends AbstractExecutorService, AbstractExecutorService implements ExecutorService
  4.         service.execute(() -> {
  5.             System.out.println("hello.word");
  6.         });
  8.         service.shutdown();
  9.     }
复制代码
3.2 newSingleThreadExecutor

上面是创建容量固定为10的线程池，还可以创建使用newSingleThreadExecutor容量为1的线程,有点单例的感觉。
  1. ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2));
复制代码
输出如下。

看看其源码。
  1.     public static ExecutorService newFixedThreadPool(int nThreads) {
  2.         return new ThreadPoolExecutor(nThreads, nThreads,
  3.                                       0L, TimeUnit.MILLISECONDS,
  4.                                       new LinkedBlockingQueue<Runnable>());
  5.     }
复制代码
里面返回的是FinalizableDelegatedExecutorService，并且其构造方法的参数是ThreadPoolExecutor对象，点进去看看到底是什么？
  1.   public static void main(String[] args) throws InterruptedException{
  2.         ExecutorService service = Executors.newSingleThreadExecutor(); //ThreadPoolExecutor extends AbstractExecutorService, AbstractExecutorService implements ExecutorService
  3.         for (int i = 0; i < 2; i++) {
  4.             service.submit(() -> {
  5.                 System.out.println(Thread.currentThread().getName() + "execute");
  6.                 try {
  7.                     TimeUnit.SECONDS.sleep(1);
  8.                 } catch (InterruptedException exception) {
  9.                     exception.printStackTrace();
  10.                 }
  11.                 System.out.println(Thread.currentThread().getName() + "finished");
  12.             });
  13.         }
  15.         service.shutdown();
  16.     }
复制代码
原来里面重写了finalize，当线程池被gc时会调用shutdown方法。将传过来的ThreadPoolExecutor又传给了其父类DelegatedExecutorService.点 super(executor)看看里面做了什么？
  1. public static ExecutorService newSingleThreadExecutor() {
  2.         return new FinalizableDelegatedExecutorService
  3.             (new ThreadPoolExecutor(1, 1,
  4.                                     0L, TimeUnit.MILLISECONDS,
  5.                                     new LinkedBlockingQueue<Runnable>()));
  6.     }
复制代码
原来最后还是调用ExecutorService来实现的逻辑呀。有没有代理的感觉？包装的目的主要是为了安全性。我们看看将它创建的对象强转成ThreadPoolExecutor。
  1. static class FinalizableDelegatedExecutorService
  2.         extends DelegatedExecutorService {
  3.         FinalizableDelegatedExecutorService(ExecutorService executor) {
  4.             super(executor);
  5.         }
  6.         protected void finalize() {
  7.             super.shutdown();
  8.         }
  9.     }
复制代码
报错。这是因为它根本不是ThreadPoolExecutor类型的对象。

如果是newFixedThreadPool强转运行就不会报错。
  1. static class DelegatedExecutorService extends AbstractExecutorService {
  2.         private final ExecutorService e;
  3.         DelegatedExecutorService(ExecutorService executor) { e = executor; }
  4.         public void execute(Runnable command) { e.execute(command); }
  5.         public void shutdown() { e.shutdown(); }
  6.         public List<Runnable> shutdownNow() { return e.shutdownNow(); }
  7.         public boolean isShutdown() { return e.isShutdown(); }
  8.         public boolean isTerminated() { return e.isTerminated(); }
  9.         ...
  10. }
复制代码
输出如下。
强转以后可以干什么，当然是动态修改参数了。比如。
  1.   public static void main(String[] args) throws InterruptedException{
  2.    ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newSingleThreadExecutor();     
  3.   }
复制代码
这样我们线程池的核心容量就改变了。想象要是newSingleThreadExecutor创建的对象也可以改变容量大小，它还可以称之为单线程线程池么？我们使用它来编码，不就有可能因为不知道其容量发生了变化出错吗？
3.3 newCachedThreadPool

看看源码，newCachedThreadPool是做什么的？
  1.   public static void main(String[] args) throws InterruptedException{
  2.         ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
  3.         }
复制代码
可以看出来，其核心线程数是0，最大线程数是Integer.MAX_VALUE，空闲等待时间是60s，等待队列没有容量。由于其最大线程数几乎可以让你想创建多少线程就创建多少线程，并且空闲线程等待时间是很长的，有60s，因此可能存在许多隐患，建议谨慎使用。
4.返回执行结果的任务

使用submit方法(而不是execute)提交任务可以获取任务执行的返回。
先使用下面带Callable传参的版本的submit
  1. public static void main(String[] args) throws InterruptedException{
  2.         ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
  3.                 executor.setCorePoolSize(10);
  4.     }
复制代码
来
  1.     public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
  2.         return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
  3.                                       60L, TimeUnit.SECONDS,
  4.                                       new SynchronousQueue<Runnable>(),
  5.                                       threadFactory);
  6.     }
复制代码
执行结果如下。

也可以使用submit的下面这个带Runnable和result重载方法。
  1.     <T> Future<T> submit(Callable<T> task);
复制代码
来。
  1. public static void main(String[] args) throws InterruptedException, ExecutionException {
  2.         ExecutorService service = Executors.newSingleThreadExecutor();
  3.         Future<String> future = service.submit(() -> "hello,world");
  4.         System.out.println(future.get()); //注意get方法不带时间参数是默认阻塞等待结果的
  6.     }
复制代码
还可以使用带FutureTask参数的构造方法。这种方式在get时是不是更加简洁一点点？
  1. <T> Future<T> submit(Runnable task, T result);
复制代码
上面三种方法都可以获取任务执行的返回结果。我们还可以通过Future对象获取当前任务的执行状态。
  1. public static void main(String[] args) throws InterruptedException, ExecutionException {
  2.         ExecutorService service = Executors.newSingleThreadExecutor();
  3.         Future<String> future = service.submit(() ->{
  4.             try {
  5.                 TimeUnit.SECONDS.sleep(1);
  6.             } catch (InterruptedException exception) {
  7.                 exception.printStackTrace();
  8.             }
  9.         }, "hello,world");
  10.         System.out.println(future.get());
  12.     }
复制代码
输出结果如下哟。
如果我们不用get来阻塞的等待执行结果，输出结果还会一样吗？
  1. public static void main(String[] args) throws InterruptedException, ExecutionException {
  2.         ExecutorService service = Executors.newSingleThreadExecutor();
  3.         FutureTask<String> task = new FutureTask(() -> "hello,world");
  4.         service.submit(task);
  5.         System.out.println(task.get());
  6.         service.shutdown();
  7.     }
复制代码
不一样了哟。

可以在任务执行过程中取消任务。
  1.     public static void main(String[] args) throws InterruptedException, ExecutionException {
  2.         ExecutorService service = Executors.newSingleThreadExecutor();
  3.         FutureTask<String> task = new FutureTask(() -> "hello,world");
  4.         service.submit(task);
  5.         System.out.println(task.get());
  6.         System.out.println(task.isDone());
  7.         System.out.println(task.isCancelled());
  8.         service.shutdown();
  9.     }
复制代码
执行结果如下。

5.执行定时任务

jdk5以后，可以使用ScheduledThreadPoolExecutor来执行定时任务，它继承自ThreadPoolExecutor。看看它的构造方法吧。可以发现其最大容量都是Integer.MAX_VALUE，并且使用的等待队列都是DelayedWorkQueue，是不是有种恍然大悟的感觉。
  1. public static void main(String[] args) throws InterruptedException, ExecutionException {
  2.         ExecutorService service = Executors.newSingleThreadExecutor();
  3.         FutureTask<String> task = new FutureTask(() -> "hello,world");
  4.         service.submit(task);
  5.         System.out.println(task.isDone());
  6.         System.out.println(task.isCancelled());
  7.         service.shutdown();
  8.     }
复制代码
来使用下吧。太简单了，是不是？
  1.   public static void main(String[] args) throws InterruptedException, ExecutionException {
  2.         ExecutorService service = Executors.newSingleThreadExecutor();
  3.         Future<String> task = service.submit(() -> {
  4.             TimeUnit.SECONDS.sleep(1);
  5.             return "hello";
  6.         });
  7.         task.cancel(true);
  8.         System.out.println(task.isDone());
  9.         System.out.println(task.isCancelled());
  10.         service.shutdown();
  11.     }
复制代码
还可以设置按照一定的频率周期来执行任务。看看源码就会用。
  1.                 public ScheduledThreadPoolExecutor(int corePoolSize) {
  2.         super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
  3.               new DelayedWorkQueue());
  4.     }
  6.    
  7.     public ScheduledThreadPoolExecutor(int corePoolSize,
  8.                                        ThreadFactory threadFactory) {
  9.         super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
  10.               new DelayedWorkQueue(), threadFactory);
  11.     }
  13.     public ScheduledThreadPoolExecutor(int corePoolSize,
  14.                                        RejectedExecutionHandler handler) {
  15.         super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
  16.               new DelayedWorkQueue(), handler);
  17.     }
  19.     public ScheduledThreadPoolExecutor(int corePoolSize,
  20.                                        ThreadFactory threadFactory,
  21.                                        RejectedExecutionHandler handler) {
  22.         super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
  23.               new DelayedWorkQueue(), threadFactory, handler);
  24.     }
复制代码
它还有一个和上面方法很相似的方法scheduleWithFixedDelay。它们甚至连参数都是一毛一样。你运行下看看，会发现好像效果也是一样的。
  1. public static void main(String[] args) {
  2.         ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1); //核心线程数1
  3.         executor.schedule(() ->System.out.println("hello,world!"), 3, TimeUnit.SECONDS);
  4.         executor.shutdown();
  5.     }
复制代码
那它们到底有什么其区别呢？您再运行下下面的代码。
  1. public static void main(String[] args) throws InterruptedException {
  2.         ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
  3.         executor.scheduleAtFixedRate(() ->System.out.println("hello,world!"), 1,3, TimeUnit.SECONDS); //初始delay1s，之后每3s执行一次
  4.         TimeUnit.SECONDS.sleep(20);
  5.         executor.shutdown();
  6.     }
复制代码
原来scheduleAtFixedRate是按一定的频率去执行任务，不管任务有没有执行完，而scheduleWithFixedDelay是在任务执行结束的基础上再delay一定时间去执行任务。

来源：https://blog.csdn.net/qq_41708993/article/details/124834799
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4