草庐IT

并发专题-Executor源码详解

pq217 2023-03-28 原文

Runnable && Thread

Runnable和Thread都是java.lang包最基本的线程操作类,相当于官方的,而Executor接口及其实现都是Doug Lea写的java.util.concurrent包下,属于民间的,当然因为太牛逼了所以也在jdk中

先看官方提供的线程操作,其中Runnable是一个函数式接口

@FunctionalInterface
public interface Runnable {
    public abstract void run();
}

可以理解为一个待执行的函数,或者理解为一个任务(通过调用run方法可以实际的执行任务)

Runable是一个定义的任务,而Thread是它的一个执行者,它提供start方法可以开启一个新线程执行传入的Runable任务,这很像命令模式,用户通过实现Runable制定一个命令,交给Thread这个执行者去具体执行

Runnable && Thread

所以一般开启新线程执行方法的方式如下

Runnable task = () -> {
    // do something
};
new Thread(task).start(); // 开启新线程执行

而开启新线程执行方法也只有这一个途径可走,就是必须通过官方的Thread.start方法

Executor

虽然开启线程执行任务只能走Thread.start方法,方法只有一个,但我们能做的是可以改变任务运行的方式,比如我们可以决定什么时候执行任务,多少个任务共用某个线程排队工作

最具代表性的就是线程池,线程池只是修改了任务执行的方式:即所有任务共用固定数量的线程,但最终的运行终归还是通过Thread.start方法实际在线程中执行Runnable方法

Doug Lea所写的java.util.concurrent.Executor把各种方式的Runnable执行器的一个抽象

public interface Executor {
    void execute(Runnable command);
}
Executor

而且提供了一些常用的执行器供我们使用,比如ThreadPoolExecutor(线程池),ForkJoinPool,当然我们也可以自己定义一个Executor按照自己的方式执行任务,比如netty中实现的SingleThreadEventExecutor是一种单个线程依次处理所有任务的执行器

ExecutorService

如果说Executor是一种对按自己套路执行任务的执行器,是一种抽象分类,那么ExecutorService就是其下的一个子分类,它是一种特殊的执行器,从名字直译来看:"执行器服务",从一个执行器升级为执行服务,像不像某公司从卖产品业务升级为产品安装售后整套服务

所以ExecutorService作为一个特殊的服务类Executor,不光能按照自己的方式执行任务,还推出了一系列附加"服务",那就看看这种特殊的执行器都提供了什么服务

public interface ExecutorService extends Executor {

    void shutdown();
    
    Future<?> submit(Runnable task);
    
    <T> Future<T> submit(Callable<T> task);
    
    ......
}

只贴了些重要方法,首先继承了Executor肯定是要继承void execute(Runnable) 方法代表它首先是一个任务执行器

shutdown方法代表这个执行器是有状态的,可以关闭服务的

然后就是重量级的submit方法,这也是ExecutorService提供的最具特色的服务,如果打开ExecutorService的类,注释第一句就写着:

/**
 * An {@link Executor} that provides methods to manage termination and
 * methods that can produce a {@link Future} for tracking progress of
 * one or more asynchronous tasks.

翻译过来大致就是ExecutorService是一个特殊的Executor执行器,他可以终止服务并且可以创建一个Future来跟踪任务执行进度

ExecutorService的submit不光能接受Runnable,还可以接受一种新型任务形式:Callable,即有返回结果和异常的任务

Callable

上面我们总结Runnable是一种可执行任务,而这种任务是没有返回结果的,也不能抛出异常,很显然现实中很多任务是需要有返回结果的,比如计算1+1等于几的任务,所以为了扩展任务类型,Doug Lea又定义一种新的任务:Callable,而ExecutorService可以接受并处理这样的任务

@FunctionalInterface
public interface Callable<V> {
    V call() throws Exception;
}
Future

submit执行的返回值是一个Future,从注释看出他可以跟踪任务的进度,它就好比任务的一个订单,通过订单可以取消任务,查看任务进度等

  • boolean cancel(boolean) 取消
    取消任务,就好比在某宝买了个东西,本质就是提交一个"把东西给我送过来"的任务,而通过订单我们就可以取消这个任务
  • isCancelled() && isDone()
    查看任务状态,是否取消和是否完成
  • V get()
    获取任务执行结果,如果没完成则阻塞,如果是Runnable,返回的就是null
ExecutorService

AbstractExecutorService

ExecutorService制订了一种新型执行器,它的特殊在于可以跟踪任务进度甚至取消任务,那么如何实现呐

首先execute方法只会单纯的执行任务,按照自己执行器的逻辑调用Thread.start方法,不会有返回值,也不支持跟踪进度或取消

所以解决方案只有一个:调包任务,具体这样操作:当用户提交任务,不是直接去execute执行,而是把任务包装为一个新任务,新任务执行原任务的同时,还负责获取任务结果,跟踪任务状态等工作,相当于是原任务的一个代理

以上即是AbstractExecutorService负责的工作,它是ExecutorService关于任务跟踪业务的相关实现(解决方案即是代理任务),继承了AbstractExecutorService的任务执行器既可以实现跟踪任务的功能

AbstractExecutorService

来看一下AbstractExecutorService的submit方法

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    // 生成一个依赖于原任务的新任务
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    // 执行新任务
    execute(ftask);
    // 返回新任务(充当任务跟踪器)
    return ftask;
}

其中newTaskFor方法负责生成新任务,同时也是原任务的跟踪器(订单)

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}

所以整个执行过程就是把Runnable或Callable转换为FutureTask的过程,而FutureTask首先是一个新任务(继承Runnable),又是原任务的跟踪器(继承Future),这种任务归类为RunnableFuture

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}
FutureTask

接下来就看FutureTask是如何实现跟踪替换原任务的,首先它的重点属性如下

  • int state; 存储原任务的执行状态
  • Callable<V> callable; 原任务Runnable也可以适配为返回null的Callable
  • Object outcome; 原任务的返回结果

再看一下重点方法

1.初始化,存储原任务,任务状态为NEW

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;       // ensure visibility of callable
}

2.run(),真实的被执行任务,加入状态判断,执行原任务通过try-catch获取异常,通过outcome保存结果

public void run() {
    // 如果任务不是新状态,直接返回
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        // 获取原任务
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                // 执行原任务
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                // 有异常设置异常
                setException(ex);
            }
            if (ran)
                // 设置outcome存储返回值
                set(result);
        }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

3.get(),获取结果,如果状态未完成阻塞等待,完成则返回outcome

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    // 如果任务未完成,阻塞等待
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    // 任务完成,返回结果    
    return report(s);
}

private V report(int s) throws ExecutionException {
    Object x = outcome;
    if (s == NORMAL)
        return (V)x; // 返回结果即outcome
    if (s >= CANCELLED)
        throw new CancellationException();
    throw new ExecutionException((Throwable)x);
}

4.cancel(),只要把状态设置为CANCEL,run时就会直接return而不会执行原任务

public boolean cancel(boolean mayInterruptIfRunning) {
    // 状态变为CANCELLED
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    // 如果正在运行调用interrupt阻断正在执行的任务
    try {    // in case call to interrupt throws exception
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();
            } finally { // final state
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}
总结

AbstractExecutorService只是实现了让任务变得可跟踪,通过调包任务,而具体任务的执行最终依然会调用execute,这个方法AbstractExecutorService并没有实现,因为这也是所有执行器的差异所在,即按自己的方式选择线程执行任务, 比如ThreadPoolExecutor线程池的固定线程数执行所有任务的模式,也就是只需要实现execute方法,而submit则交给父类AbstractExecutorService处理

有关并发专题-Executor源码详解的更多相关文章

  1. UE4 源码阅读:从引擎启动到Receive Begin Play - 2

    一、引擎主循环UE版本:4.27一、引擎主循环的位置:Launch.cpp:GuardedMain函数二、、GuardedMain函数执行逻辑:1、EnginePreInit:加载大多数模块int32ErrorLevel=EnginePreInit(CmdLine);PreInit模块加载顺序:模块加载过程:(1)注册模块中定义的UObject,同时为每个类构造一个类默认对象(CDO,记录类的默认状态,作为模板用于子类实例创建)(2)调用模块的StartUpModule方法2、FEngineLoop::Init()1、检查Engine的配置文件找出使用了哪一个GameEngine类(UGame

  2. ruby-on-rails - 获取并发布相同匹配项的请求 - 2

    在我的路线文件中我有:match'graphs/(:id(/:action))'=>'graphs#(:action)'如果是GET请求(工作)或POST请求(不工作),我想匹配它我知道我可以使用以下方法在资源中声明POST请求:post'/'=>:show,:on=>:member但是我怎样才能为比赛做到这一点呢?谢谢。 最佳答案 如果你同时想要POST和GETmatch'graphs/(:id(/:action))'=>'graphs#(:action)',:via=>[:get,:post]编辑默认值可以设置如下match'g

  3. elasticsearch源码关于TransportSearchAction【阶段三】 - 2

    1.回顾.TransportServicepublicclassTransportServiceextendsAbstractLifecycleComponentTransportService:方法:1publicfinalTextendsTransportResponse>voidsendRequest(finalTransport.Connectionconnection,finalStringaction,finalTransportRequestrequest,finalTransportRequestOptionsoptions,TransportResponseHandlerT>

  4. 物联网MQTT协议详解 - 2

    一、什么是MQTT协议MessageQueuingTelemetryTransport:消息队列遥测传输协议。是一种基于客户端-服务端的发布/订阅模式。与HTTP一样,基于TCP/IP协议之上的通讯协议,提供有序、无损、双向连接,由IBM(蓝色巨人)发布。原理:(1)MQTT协议身份和消息格式有三种身份:发布者(Publish)、代理(Broker)(服务器)、订阅者(Subscribe)。其中,消息的发布者和订阅者都是客户端,消息代理是服务器,消息发布者可以同时是订阅者。MQTT传输的消息分为:主题(Topic)和负载(payload)两部分Topic,可以理解为消息的类型,订阅者订阅(Su

  5. (附源码)vue3.0+.NET6实现聊天室(实时聊天SignalR) - 2

    参考文章搭建文章gitte源码在线体验可以注册两个号来测试演示图:一.整体介绍  介绍SignalR一种通讯模型Hub(中心模型,或者叫集线器模型),调用这个模型写好的方法,去发送消息。  内容有:    ①:Hub模型的方法介绍    ②:服务器端代码介绍    ③:前端vue3安装并调用后端方法    ④:聊天室样例整体流程:1、进入网站->调用连接SignalR的方法2、与好友发送消息->调用SignalR的自定义方法 前端通过,signalR内置方法.invoke()  去请求接口3、监听接受方法(渲染消息)通过new signalR.HubConnectionBuilder().on

  6. Tcl脚本入门笔记详解(一) - 2

    TCL脚本语言简介•TCL(ToolCommandLanguage)是一种解释执行的脚本语言(ScriptingLanguage),它提供了通用的编程能力:支持变量、过程和控制结构;同时TCL还拥有一个功能强大的固有的核心命令集。TCL经常被用于快速原型开发,脚本编程,GUI和测试等方面。•实际上包含了两个部分:一个语言和一个库。首先,Tcl是一种简单的脚本语言,主要使用于发布命令给一些互交程序如文本编辑器、调试器和shell。由于TCL的解释器是用C\C++语言的过程库实现的,因此在某种意义上我们又可以把TCL看作C库,这个库中有丰富的用于扩展TCL命令的C\C++过程和函数,所以,Tcl是

  7. Ruby 并发/异步处理(简单用例) - 2

    我一直在研究ruby​​的并行/异步处理能力,并阅读了许多文章和博客文章。我查看了EventMachine、Fibers、Revactor、Reia等。不幸的是,我无法为这个非常简单的用例找到简单、有效(且非IO阻塞)的解决方案:File.open('somelogfile.txt')do|file|whileline=file.gets#(R)ReadfromIOline=process_line(line)#(P)Processthelinewrite_to_db(line)#(W)WritetheoutputtosomeIO(DBorfile)endend你看到了吗,我的小脚本正

  8. ruby-on-rails - 乘客并发连接错误 - 2

    我有生产服务器(Nginx+Passenger)。当我尝试从另一台计算机ab-n3-c3myhost.ru/时,我在我的nginxerror.log中收到此错误日志:[pid=21160thr=139775297914624file=ext/nginx/HelperAgent.cpp:584time=2011-08-3115:25:49.22]:UncaughtexceptioninPassengerServerclientthread:exception:Cannotreadresponsefrombackendprocess:Connectionresetbypeer(104)ba

  9. Cesium源码解析一(terrain文件的加载、解析与渲染全过程梳理) - 2

    快速导航(持续更新中…)Cesium源码解析一(terrain文件的加载、解析与渲染全过程梳理)Cesium源码解析二(metadataAvailability的含义)Cesium源码解析三(metadata元数据拓展中行列号的分块规则解析)Cesium源码解析四(Quantized-Mesh(.terrain)格式文件在CesiumJS和UE中加载情况的对比)目录1.前言2.本篇的由来3.terrain文件的加载3.1更新环境3.2更新和执行渲染命令3.3数据优化3.4结束当前帧4.总结1.前言  目前市场上三维比较火的实现方案主要有两种,b/s的方案主要是Cesium,c/s的方案主要是u

  10. 【详解】Docker安装Elasticsearch7.16.1集群 - 2

    开门见山|拉取镜像dockerpullelasticsearch:7.16.1|配置存放的目录#存放配置文件的文件夹mkdir-p/opt/docker/elasticsearch/node-1/config#存放数据的文件夹mkdir-p/opt/docker/elasticsearch/node-1/data#存放运行日志的文件夹mkdir-p/opt/docker/elasticsearch/node-1/log#存放IK分词插件的文件夹mkdir-p/opt/docker/elasticsearch/node-1/plugins若你使用了moba,直接右键新建即可如上图所示依次类推创建

随机推荐