并发编程(六)-AbstractExecutorService源码分析

一、AbstractExecutorService简介


AbstractExecutorService是一个抽象类,实现了ExecutorService接口,提供了线程池的基本实现。它是Java Executor框架的核心类,提供了线程池的基本操作,如提交任务、管理线程池、执行任务等。

自定义执行器服务可用扩展AbreactExecutorService 并覆盖其方法以提供自己的实现。这使开发人员可以创建适合其特定需求的执行器服务。

二、AbstractExecutorService如何使用


AbstractExecutorService是一个抽象类,不能直接实例化,需要通过继承它的子类ThreadPoolExecutorScheduledThreadPoolExecutor来使用。使用AbstractExecutorService创建线程池的步骤如下:

2.1 没有返回值

// 1、创建ThreadPoolExecutor  创建一个固定大小的线程池,核心线程数为1,最大线程数为5,任务队列大小为10
ExecutorService executorService = new ThreadPoolExecutor(1, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(10));
// 2、提交任务
executorService.submit(new Runnable() {
    @Override
    public void run() {
        //执行任务
    }
});
// 3、关闭线程池
executorService.shutdown();

2.2 有返回值

// 1、创建ThreadPoolExecutor  创建一个固定大小的线程池,核心线程数为1,最大线程数为5,任务队列大小为10
ExecutorService executorService = new ThreadPoolExecutor(1, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(10));
// 2、提交一个Callable任务
Future<String> future = executorService.submit(new Callable<String>() {
    @Override
    public String call() throws Exception {
        //执行任务,返回结果
        return "hello ExecutorService";
    }
});
// 3、获取值
String result = future.get();
// 4、关闭线程池
executorService.shutdown();

总之,使用AbstractExecutorService创建线程池非常简单,只需要创建线程池对象、提交任务、关闭线程池即可。同时,AbstractExecutorService也提供了一些高级功能,如异步任务、定时任务等,可以根据具体需求选择使用。

三、如何自定义AbstractExecutorService


要自定义一个延迟AbstractExecutorService,以下步骤:

  1. 继承AbstractExecutorService类

  1. 实现 submit 方法 覆盖 submit 方法,将任务添加到队列中,其中在任务执行前应用延迟。

  1. 实现 execute 方法 覆盖 execute 方法,调用 submit 方法并返回 Future 对象,等待任务完成。

  1. 实现shutdown 方法 覆盖 shutdown 方法,停止所有任务并等待它们完成。

  1. 实现 DelayedTask 类 创建一个 DelayedTask 类,用于封装任务和延迟时间。

  1. 实现 RunnableTask 和 CallableTask 类 创建 RunnableTask 和 CallableTask 类,用于封装任务。

public class CustomDelayedExecutorService extends AbstractExecutorService {
    
    private final DelayQueue<DelayedTask<?>> queue = new DelayQueue<>();
    
    @Override
    public <T> Future<T> submit(Callable<T> task) {
        return schedule(new CallableTask<>(task));
    }
    
    @Override
    public Future<?> submit(Runnable task) {
        return schedule(new RunnableTask(task));
    }
    
    private <T> Future<T> schedule(DelayedTask<T> task) {
        queue.offer(task);
        return task;
    }
    
    @Override
    public void execute(Runnable command) {
        submit(command);
    }
    
    @Override
    public void shutdown() {
        for (DelayedTask<?> task : queue) {
            task.cancel(false);
        }
    }
    
    @Override
    public List<Runnable> shutdownNow() {
        List<Runnable> tasks = new ArrayList<>();
        for (DelayedTask<?> task : queue) {
            if (task.cancel()) {
                tasks.add(task.getTask());
            }
        }
        return tasks;
    }
    
    @Override
    public boolean isShutdown() {
        return false;
    }
    
    @Override
    public boolean isTerminated() {
        return false;
    }
    
    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return false;
    }
​
}
class DelayedTask<T> implements RunnableFuture<T>, Delayed {
    private final long delay;
    private final long expire;
    private final RunnableFuture<T> task;
    public DelayedTask(RunnableFuture<T> task, long delay) {
        this.task = task;
        this.delay = delay;
        this.expire = System.nanoTime() + TimeUnit.NANOSECONDS.convert(delay, TimeUnit.MILLISECONDS);
    }
   
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(expire - System.nanoTime(), TimeUnit.NANOSECONDS);
    }
   
    @Override
    public int compareTo(Delayed o) {
        return Long.compare(expire, ((DelayedTask<?>) o).expire);
    }
   
    @Override
    public void run() {
        if (!isCancelled()) {
            task.run();
        }
    }
   
    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return task.cancel(mayInterruptIfRunning);
    }
   
    @Override
    public boolean isCancelled() {
        return task.isCancelled();
    }
   
    @Override
    public boolean isDone() {
        return task.isDone();
    }
   
    @Override
    public T get() throws InterruptedException, ExecutionException {
        return task.get();
    }
   
    @Override
    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return task.get(timeout, unit);
    }
   
    public RunnableFuture<T> getTask() {
        return task;
    }
}
​
class RunnableTask implements RunnableFuture<Void> {
    private final Runnable task;
    public RunnableTask(Runnable task) {
        this.task = task;
    }
    @Override
    public void run() {
        task.run();
    }
    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return false;
    }
    @Override
    public boolean isCancelled() {
        return false;
    }
    @Override
    public boolean isDone() {
        return true;
    }
    @Override
    public Void get() throws InterruptedException, ExecutionException {
        return null;
    }
    @Override
    public Void get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return null;
    }
}
class CallableTask<T> extends FutureTask<T> {
    public CallableTask(Callable<T> callable) {
        super(callable);
    }
}

四、AbstractExecutorService源码分析


4.1 AbstractExecutorService dragrams

4.2 方法分析

4.2.1 newTaskFor(Runnable runnable, T value) 方法

AbstractExecutorService中的newTaskFor方法,用于创建一个RunnableFuture对象,其中RunnableFuture是Future和Runnable接口的结合体,可以用来表示异步执行的结果。该方法的参数包括一个Runnable对象和一个泛型T的值。通过传入的Runnable对象和T类型的值,创建了一个FutureTask对象,并将其返回。

FutureTask是RunnableFuture的一个具体实现,它表示一个可取消的异步计算,可以通过调用get方法获取计算结果。在创建FutureTask对象时,需要传入一个Callable对象或Runnable对象,用于执行异步计算。而在这里,我们传入了一个Runnable对象和一个T类型的值,表示该Runnable对象的执行结果为T类型的值。

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
      // 将runnable包装成FutureTask类型
      return new FutureTask<T>(runnable, value);
 }

4.2.2 submit(Runnable task)方法

AbstractExecutorService中的submit方法是ExecutorService接口中的一个方法,用于提交一个Runnable任务,并返回一个表示该任务异步执行结果的Future对象。

  1. 如果传入的task为null,则抛出NullPointerException异常;

  1. 调用newTaskFor(Runnable runnable, T value)方法创建一个RunnableFuture对象ftask,其中RunnableFuture是Future和Runnable接口的结合体,用于表示异步执行的结果。在该方法中,我们将泛型T的值设置为null,表示该Runnable任务没有返回值;

  1. 通过调用execute(Runnable command)方法提交任务进行执行,该方法是一个抽象方法,需要由子类实现。在该方法中,我们将创建的ftask对象作为参数传入,表示需要执行的任务为ftask对象。

  1. 最后,将ftask对象返回,作为Future对象,用于获取任务执行结果或取消任务。

public Future<?> submit(Runnable task) {
      // 如果任务为null,则抛出nullPointerException
      if (task == null) throw new NullPointerException();
      // 将task包装成RunnableFuture
      RunnableFuture<Void> ftask = newTaskFor(task, null);
      // 通过调用execute(Runnable command)方法提交任务进行执行
      execute(ftask);
      // 返回RunnableFuture
      return ftask;
 }

4.2.3 doInvokeAny(Collection<? extends Callable<T>> tasks,boolean timed, long nanos)方法

AbstractExecutorService中doInvokeAny方法用于执行一个任务集合中的所有任务并返回第一个成功执行的任务的结果,如果所有任务都执行失败则抛出ExecutionException异常。其中,方法参数tasks为任务集合,timed和nanos用于指定超时时间,如果超时则抛出TimeoutException异常。

  1. 如果任务集合是否为空,则抛出NullPointerException异常

  1. 如果任务集合的大小为0,则抛出IllegalArgumentException异常

  1. futures用于存储所有任务的Future对象, ecs对象用于异步执行任务集合中的任务。

  1. 从迭代器中获取一个任务xecutorCompletionService,并将该任务的执行结果的Future添加到futures列表中

  1. 死循环不断从ExecutorCompletionService中获取已完成的任务的结果Future,直到有一个任务完成为止,具体细流程如下:

  • 调用ExecutorCompletionService的poll()方法获取一个已完成的任务的结果Future,如果poll()返回null,表示当前没有已完成的任务。

  • 当前没有已完成的任务

  • 还有未执行的任务(ntasks > 0),则从任务集合中获取下一个任务,并将其提交给ExecutorCompletionService执行。提交成功后,活跃线程数active加一;

  • 没有执行的任务,并且活跃线程数为0(即所有任务都已执行完毕),则退出循环;

  • 指定超时时间(timed为true),则调用poll(nanos, TimeUnit.NANOSECONDS)方法获取已完成的任务的结果Future,并等待最多nanos纳秒的时间。如果超时则抛出TimeoutException;

  • 没有指定超时时间,则调用take()方法阻塞等待已完成的任务的结果Future。

  • 如果获取到了已完成的任务的结果Future,则将活跃线程数active减一,并尝试获取该任务的执行结果。

  • 如果获取结果时出现异常,则将异常保存在ExecutionException中,继续下一轮循环;

  • 如果成功获取到了一个任务的执行结果,则直接返回该结果。

  1. 如果所有任务都已经完成,但是没有找到任何一个成功的任务,则抛出ExecutionException异常。

  1. 最后,取消所有尚未完成的任务,以便节省资源并提高效率

private<T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
                            boolean timed, long nanos)
      throws InterruptedException, ExecutionException, TimeoutException {
     // 如果任务集合是否为空,则抛出NullPointerException异常
      if (tasks == null)
          throw new NullPointerException();
      int ntasks = tasks.size();
      // 如果任务集合的大小为0,则抛出IllegalArgumentException异常
      if (ntasks == 0)
          throw new IllegalArgumentException();
      // futures用于存储所有任务的Future对象   
      ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
      ExecutorCompletionService<T> ecs =
          new ExecutorCompletionService<T>(this);
      try {
          ExecutionException ee = null;
          // 计算deadline时间
          final long deadline = timed ? System.nanoTime() + nanos : 0L;
          Iterator<? extends Callable<T>> it = tasks.iterator();
          // 从迭代器中获取一个任务xecutorCompletionService,并将该任务的执行结果的Future添加到futures列表中。
          futures.add(ecs.submit(it.next()));
          // 任务数减1
          --ntasks;
          // 退出循环标识(也就是调用ecs.sumbit的任务数)
          int active = 1;
          // 自旋、死循环
          for (;;) {
              // 获取Future对象
              Future<T> f = ecs.poll();
              // 判断future 任务是否已完成
              if (f == null) {
                 // 任务没有完成,继续判断是否还有未执行的任务
                  if (ntasks > 0) {
                      // 任务s数减1
                      --ntasks;
                      // 再从tasks中获取一个任务,提交到ExecutorCompletionService中进行执行。
                      futures.add(ecs.submit(it.next()));
                     // 提交ecs中加1
                      ++active;
                  }
                  else if (active == 0)
                      break;
                  // 有超时时间限制    
                  else if (timed) {
                  //调用ExecutorCompletionService的poll(long timeout, TimeUnit unit)方法等待指定的超时时间。
                      f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                     // 在等待过程中已经超过了指定的超时时间,因此会抛出 TimeoutException 异常
                      if (f == null)
                          throw new TimeoutException();
                      // 将剩余的超时时间重新计算,并继续执行后续的代码。
                      nanos = deadline - System.nanoTime();
                  }
                  // 如果没有超时时间限制,则调用ExecutorCompletionService的take()方法一直等待,直到有任务执行完成。
                  else
                      f = ecs.take(); 
              }
              // 任务已完成(获取到Future对象)
              if (f != null) {
                  --active;
                  try {
                    // 尝试获取Future对象的执行结果,
                      return f.get();
                  } catch (ExecutionException eex) {
                  // 如果获取失败,则将异常保存在ee变量中,继续等待下一个任务的执行结果
                      ee = eex;
                  } catch (RuntimeException rex) {
                      ee = new ExecutionException(rex);
                  }
              }
          }
          // 如果所有任务都已经完成,但是没有找到任何一个成功的任务,则抛出ExecutionException异常。
          if (ee == null)
              ee = new ExecutionException();
          throw ee;

      } finally {
         // 取消所有尚未完成的任务,以便节省资源并提高效率
          for (int i = 0, size = futures.size(); i < size; i++)
              futures.get(i).cancel(true);
      }
  }

4.2.4 invokeAll(Collection<? extends Callable<T>> tasks)方法

AbstractExecutorService中的invokeAll方法,它的作用是在执行给定的任务集合tasks中的所有任务,并等待所有任务完成后返回一个包含Future对象的列表,Future对象可以用来获取每个任务的执行结果。如果其中某个任务抛出异常,则该异常将传播到调用者。

  • 判断任务是否为空,如果空,则抛出NullPointerException异常;

  • 遍历参数tasks中的每个Callable任务,将每个任务转换成一个RunnableFuture对象,并把任务添加到futures中,然后调用execute方法执行任务。

  • 循环遍历futures中的每个Future对象,如果某个Future对象的任务还没有完成,则调用get()方法等待任务完成,如果任务抛出异常,则忽略异常。

  • 如果所有的任务都成功完成,则将done标记为true,并返回futures;

  • 否则,取消所有未完成的任务,并抛出InterruptedException异常。

public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
    throws InterruptedException {
    // 判断任务是否为空,如果空,则抛出NullPointerException异常;
    if (tasks == null)
        throw new NullPointerException();
    ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
    boolean done = false;
    try {
        // 遍历参数tasks中的每个Callable任务
        for (Callable<T> t : tasks) {
            // 将任务转换成一个RunnableFuture对象
            RunnableFuture<T> f = newTaskFor(t);
            futures.add(f);
            // 调用execute方法执行任务
            execute(f);
        }
        // 循环遍历futures中的每个Future对象
        for (int i = 0, size = futures.size(); i < size; i++) {
            Future<T> f = futures.get(i);
            // 如果某个Future对象的任务还没有完成
            if (!f.isDone()) {
                try {
                    // 调用get()方法等待任务完成,如果任务抛出异常,则忽略异常。
                    f.get();
                } catch (CancellationException ignore) {
                } catch (ExecutionException ignore) {
                }
            }
        }
        // 如果所有的任务都成功完成,则将done标记为true,并返回futures;
        done = true;
        return futures;
    } finally {
        if (!done)
           // 如果所有的任务没有全部成功完成,取消所有未完成的任务
            for (int i = 0, size = futures.size(); i < size; i++)
                futures.get(i).cancel(true);
    }
}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mfbz.cn/a/4423.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

阻塞队列(BlockingQueue)的实现和使用

阻塞队列&#xff08;BlockingQueue&#xff09; 文章目录阻塞队列&#xff08;BlockingQueue&#xff09;阻塞队列的梗概解耦合和削峰填谷java代码实现一个阻塞队列阻塞队列的梗概 众所周知&#xff0c;队列是一种数据结构&#xff0c;符合先进先出的结构&#xff0c;先进先出…

【动态绘图】python可视化--丝滑版

✅作者简介&#xff1a;在读博士&#xff0c;伪程序媛&#xff0c;人工智能领域学习者&#xff0c;深耕机器学习&#xff0c;交叉学科实践者&#xff0c;周更前沿文章解读&#xff0c;提供科研小工具&#xff0c;分享科研经验&#xff0c;欢迎交流&#xff01;&#x1f4cc;个人…

鼎桥通信,拥抱基础创新的“高灵活性”时代

作者 | 曾响铃 文 | 响铃说 伴随数智化转型成为时代变革大方向&#xff0c;一批走在时代前端的数智化转型企业应运而生&#xff0c;不断丰富5G、物联网等新兴技术的应用场景&#xff0c;构建万智互联的产业生态。作为国内通信领域的引领者&#xff0c;鼎桥通信技术有限公司&a…

AF染料试剂Alexa fluor 680 PEG Biotin,AF680 PEG Biotin,荧光强度稳定利于多种荧光标记

文章关键词&#xff1a;AF染料试剂&#xff0c;AF680&#xff0c;PE-Biotin衍生物Alexa fluor 680 PEG Biotin&#xff0c;AF680 PEG Biotin | Alexa fluor 680-PEG-生物素| CAS&#xff1a;N/A | 纯度&#xff1a;95%试剂参数信息&#xff1a; CAS&#xff1a;N/A 外观&am…

docker使用

dokcer 安装 # 1、yum 包更新到最新 yum update # 2、安装需要的软件包&#xff0c; yum-util 提供yum-config-manager功能&#xff0c;另外两个是devicemapper驱动依赖的 yum install -y yum-utils device-mapper-persistent-data lvm2 # 3、 设置yum源 yum-config-manager …

精确性和准确性是两码事儿

准确性(Accuracy)是与正确答案的接近程度&#xff0c;精确性(Precision)是对这个答案的分辨率。 假设&#xff0c;你问我&#xff0c;”现在几点了?” 我抬头看看太阳&#xff0c;然后估算了一下&#xff0c;回答道 “现在是上午 10 点 35 分 22.131 秒” 我给出的是一个足…

Nacos配置中心优雅配置JSON数据格式

在我业务开发中&#xff0c;需要在配置中心配置Json数据&#xff0c;返回给前端。因Nacos默认不支持Json格式配置&#xff0c;需要搭配监听器获取配置中心Json数据&#xff0c;返回给客户端。二、搭配Nacos配置Josn数据1. bootstrap.ymlserver:port: 9000 spring:application:n…

Vue使用ElementUI对table的指定列进行合算

前言 最近有一个想法&#xff0c;就是记录自己花销的时候&#xff0c;table中有一项内容是花销的金额。然后想在table的底部有一项内容是该金额的总计。 然后我就顺着elementui的table组件寻找相关的demo&#xff0c;还真发现了一个这样的demo。 对于这个demo&#xff0c;官方…

SSH框架整合教程

工程目录结构如下&#xff1a; 本工程只介绍SSH整合的基本流程&#xff0c;所以没有写接口 1. 导入jar包 <dependencies><!--hibernate包--><dependency><groupId>org.hibernate</groupId><artifactId>hibernate-core</artifactId>…

【各种安装2】

各种安装2一、八阶段-第四章-案例导入说明1.安装MySQL1.1.准备目录1.2.运行命令1.3.修改配置1.4.重启2.导入SQL3.导入Demo工程3.1.分页查询商品3.2.新增商品3.3.修改商品3.4.修改库存3.5.删除商品3.6.根据id查询商品3.7.根据id查询库存3.8.启动4.导入商品查询页面4.1.运行nginx…

Linux线程同步与互斥(二)/生产消费者模型

⭐前言&#xff1a;本文会先后讲解生产消费者模型、条件变量和基于阻塞队列的生产消费者模型。 1.生产消费者模型 什么是生产消费者模型&#xff1f; 认识生产消费者模型 使用学生&#xff08;消费者&#xff09;&#xff0c;超市&#xff0c;供货商&#xff08;生产者&…

C++ 26 常用算法

目录 一、概述 1.1 常用遍历算法 1.1.1 算法简介 1.1.2 for_each遍历算法 1.1.3 transform遍历算法 1.2 常用查找算法 1.2.1 算法简介 1.2.2 find 查找算法 1.2.3 find_if 查找算法 1.2.4 adjacent_find 查找算法 1.2.5 binary_search 查找算法 1.2.6 count 查找算法…

【面试题】JS的一些优雅写法 reduce和map

大厂面试题分享 面试题库 前后端面试题库 &#xff08;面试必备&#xff09; 推荐&#xff1a;★★★★★ 地址&#xff1a;前端面试题库 web前端面试题库 VS java后端面试题库大全 JS的一些优雅写法 reduce 1、可以使用 reduce 方法来实现对象数组中根据某一key值求和 …

LFM雷达实现及USRP验证【章节3:连续雷达测距测速】

第一章介绍了在相对速度为0时候的雷达测距原理 目录 1. LFM测速 1.1 雷达测速原理 1.2 Chrip信号测速 2. LFM测速代码实现 参数设置 仿真图像 matlab源码 代码分析 第一章介绍了在相对速度为0时候的雷达测距原理&#xff0c;第二章介绍了基于LFM的雷达测距原理及其实现…

数据结构第十一期——线段树的原理和应用

目录 一、前言 二、线段树的概念 1、区间最值问题RMQ (Range Minimum/Maximum Query) &#xff08;1&#xff09;暴力法 &#xff08;2&#xff09;高效的办法&#xff1a;线段树 &#xff08;3&#xff09;把数列放在二叉树上 &#xff08;4&#xff09;查询最小值的复…

43-二叉树练习-LeetCode236二叉树的最近公共祖先

题目 给定一个二叉树, 找到该树中两个指定节点的最近公共祖先。 百度百科中最近公共祖先的定义为&#xff1a;“对于有根树 T 的两个节点 p、q&#xff0c;最近公共祖先表示为一个节点 x&#xff0c;满足 x 是 p、q 的祖先且 x 的深度尽可能大&#xff08;一个节点也可以是它…

一款全新的基于GPT4的Python神器,关键还免费

chartgpt大火之后&#xff0c;随之而来的就是一大类衍生物了。 然后&#xff0c;今天要给大家介绍的是一款基于GPT4的新一代辅助编程神器——Cursor。 它最值得介绍的地方在于它免费&#xff0c;我们可以直接利用它来辅助我们编程&#xff0c;真正做到事半功倍。 注意&#…

k8s实践 | configmapsecretpvpvc

文章目录configmap&secret&pv&pvc一、configMap1、应用场景2、创建configMap2.1、help文档2.2、使用目录创建2.3、根据文件创建2.4、文字创建2.5、直接方法2.6、pod中应用2.7、热更新二、secret1、Service Account2、opaque Secret2.1、创建示例2.2、使用方式三、k…

eNSP 本地AAA配置实验

关于本实验本实验要求将路由器AR1配置为AAA服务器&#xff0c;以本地认证方式对尝试登录AR1的用户进行身份认证和授权。路由器AR2作为登录用户&#xff08;AAA客户端&#xff09;&#xff0c;以Telnet的方式登录AR1.读者需要在AR1中创建一个名为datacom的管理员域&#xff0c;并…

【Unity游戏开发教程】零基础带你从小白到超神22——旧动画和新动画组件的使用

制作一个动画 创建动画 添加变化属性 实现方块向右移动10 添加关键帧 实现先慢后快的效果 录制动画 旧动画组件(Animation组件) 如果想让一个游
最新文章