找回密码
 立即注册
首页 业界区 业界 线程池中execute和submit的区别?

线程池中execute和submit的区别?

聱嘹 2025-6-12 09:46:10
简要回答

execute只能提交Runnable类型的任务,无返回值。submit既可以提交Runnable类型的任务,也可以提交Callable类型的任务,会有一个类型为Future的返回值,但当任务类型为Runnable时,返回值为null。
execute在执行任务时,如果遇到异常会直接抛出,而submit不会直接抛出,只有在使用Future的get方法获取返回值时,才会抛出异常。
execute所属顶层接口是Executor,submit所属顶层接口是ExecutorService,实现类ThreadPoolExecutor重写了execute方法,抽象类AbstractExecutorService重写了submit方法。
扩展知识

通过执行execute方法 该方法无返回值,为ThreadPoolExecutor自带方法,传入Runnable类型对象
1.png


  • 通过执行submit方法 该方法返回值为Future对象,为抽象类AbstractExecutorService的方法,被ThreadPoolExecutor继承,其内部实现也是调用了接口类Executor的execute方法,通过上面的类图可以看到,该方法的实现依然是ThreadPoolExecutor的execute方法
2.png

3.png

execute()执行流程图

4.png

execute()源码
  1.     // 使用原子操作类AtomicInteger的ctl变量,前3位记录线程池的状态,后29位记录线程数
  2.     private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
  3.     // Integer的范围为[-2^31,2^31 -1], Integer.SIZE-3 =32-3= 29,用来辅助左移位运算
  4.     private static final int COUNT_BITS = Integer.SIZE - 3;
  5.     // 高三位用来存储线程池运行状态,其余位数表示线程池的容量
  6.     private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
  7.     // 线程池状态以常量值被存储在高三位中
  8.     private static final int RUNNING    = -1 << COUNT_BITS; // 线程池接受新任务并会处理阻塞队列中的任务
  9.     private static final int SHUTDOWN   =  0 << COUNT_BITS; // 线程池不接受新任务,但会处理阻塞队列中的任务
  10.     private static final int STOP       =  1 << COUNT_BITS; // 线程池不接受新的任务且不会处理阻塞队列中的任务,并且会中断正在执行的任务
  11.     private static final int TIDYING    =  2 << COUNT_BITS; // 所有任务都执行完成,且工作线程数为0,将调用terminated方法
  12.     private static final int TERMINATED =  3 << COUNT_BITS; // 最终状态,为执行terminated()方法后的状态
  13.     // ctl变量的封箱拆箱相关的方法
  14.     private static int runStateOf(int c)     { return c & ~CAPACITY; } // 获取线程池运行状态
  15.     private static int workerCountOf(int c)  { return c & CAPACITY; } // 获取线程池运行线程数
  16.     private static int ctlOf(int rs, int wc) { return rs | wc; } // 获取ctl对象
复制代码
  1. public void execute(Runnable command) {
  2.     if (command == null) // 任务为空,抛出NPE
  3.         throw new NullPointerException();
  4.         
  5.     int c = ctl.get(); // 获取当前工作线程数和线程池运行状态(共32位,前3位为运行状态,后29位为运行线程数)
  6.     if (workerCountOf(c) < corePoolSize) { // 如果当前工作线程数小于核心线程数
  7.         if (addWorker(command, true)) // 在addWorker中创建核心线程并执行任务
  8.             return;
  9.         c = ctl.get();
  10.     }
  11.    
  12.     // 核心线程数已满(工作线程数>核心线程数)才会走下面的逻辑
  13.     if (isRunning(c) && workQueue.offer(command)) { // 如果当前线程池状态为RUNNING,并且任务成功添加到阻塞队列
  14.         int recheck = ctl.get(); // 双重检查,因为从上次检查到进入此方法,线程池可能已成为SHUTDOWN状态
  15.         if (! isRunning(recheck) && remove(command)) // 如果当前线程池状态不是RUNNING则从队列删除任务
  16.             reject(command); // 执行拒绝策略
  17.         else if (workerCountOf(recheck) == 0) // 当线程池中的workerCount为0时,此时workQueue中还有待执行的任务,则新增一个addWorker,消费workqueue中的任务
  18.             addWorker(null, false);
  19.     }
  20.     // 阻塞队列已满才会走下面的逻辑
  21.     else if (!addWorker(command, false)) // 尝试增加工作线程执行command
  22.         // 如果当前线程池为SHUTDOWN状态或者线程池已饱和
  23.         reject(command); // 执行拒绝策略
  24. }
复制代码
  1. private boolean addWorker(Runnable firstTask, boolean core) {
  2.     retry: // 循环退出标志位
  3.     for (;;) { // 无限循环
  4.         int c = ctl.get();
  5.         int rs = runStateOf(c); // 线程池状态
  6.         // Check if queue empty only if necessary.
  7.         if (rs >= SHUTDOWN &&
  8.             ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()) // 换成更直观的条件语句
  9.             // (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())
  10.            )
  11.            // 返回false的条件就可以分解为:
  12.            //(1)线程池状态为STOP,TIDYING,TERMINATED
  13.            //(2)线程池状态为SHUTDOWN,且要执行的任务不为空
  14.            //(3)线程池状态为SHUTDOWN,且任务队列为空
  15.             return false;
  16.         // cas自旋增加线程个数
  17.         for (;;) {
  18.             int wc = workerCountOf(c); // 当前工作线程数
  19.             if (wc >= CAPACITY ||
  20.                 wc >= (core ? corePoolSize : maximumPoolSize)) // 工作线程数>=线程池容量 || 工作线程数>=(核心线程数||最大线程数)
  21.                 return false;
  22.             if (compareAndIncrementWorkerCount(c)) // 执行cas操作,添加线程个数
  23.                 break retry; // 添加成功,退出外层循环
  24.             // 通过cas添加失败
  25.             c = ctl.get();  
  26.             // 线程池状态是否变化,变化则跳到外层循环重试重新获取线程池状态,否者内层循环重新cas
  27.             if (runStateOf(c) != rs)
  28.                 continue retry;
  29.             // else CAS failed due to workerCount change; retry inner loop
  30.         }
  31.     }
  32.     // 简单总结上面的CAS过程:
  33.     //(1)内层循环作用是使用cas增加线程个数,如果线程个数超限则返回false,否者进行cas
  34.     //(2)cas成功则退出双循环,否者cas失败了,要看当前线程池的状态是否变化了
  35.     //(3)如果变了,则重新进入外层循环重新获取线程池状态,否者重新进入内层循环继续进行cas
  36.     // 走到这里说明cas成功,线程数+1,但并未被执行
  37.     boolean workerStarted = false; // 工作线程调用start()方法标志
  38.     boolean workerAdded = false; // 工作线程被添加标志
  39.     Worker w = null;
  40.     try {
  41.         w = new Worker(firstTask); // 创建工作线程实例
  42.         final Thread t = w.thread; // 获取工作线程持有的线程实例
  43.         if (t != null) {
  44.             final ReentrantLock mainLock = this.mainLock; // 使用全局可重入锁
  45.             mainLock.lock(); // 加锁,控制并发
  46.             try {
  47.                 // Recheck while holding lock.
  48.                 // Back out on ThreadFactory failure or if
  49.                 // shut down before lock acquired.
  50.                 int rs = runStateOf(ctl.get()); // 获取当前线程池状态
  51.                 // 线程池状态为RUNNING或者(线程池状态为SHUTDOWN并且没有新任务时)
  52.                 if (rs < SHUTDOWN ||
  53.                     (rs == SHUTDOWN && firstTask == null)) {
  54.                     if (t.isAlive()) // 检查线程是否处于活跃状态
  55.                         throw new IllegalThreadStateException();
  56.                     workers.add(w); // 线程加入到存放工作线程的HashSet容器,workers全局唯一并被mainLock持有
  57.                     int s = workers.size();
  58.                     if (s > largestPoolSize)
  59.                         largestPoolSize = s;
  60.                     workerAdded = true;
  61.                 }
  62.             } finally {
  63.                 mainLock.unlock(); // finally块中释放锁
  64.             }
  65.             if (workerAdded) { // 线程添加成功
  66.                 t.start(); // 调用线程的start()方法
  67.                 workerStarted = true;
  68.             }
  69.         }
  70.     } finally {
  71.         if (! workerStarted) // 如果线程启动失败,则执行addWorkerFailed方法
  72.             addWorkerFailed(w);
  73.     }
  74.     return workerStarted;
  75. }
复制代码
Worker源码

Worker是ThreadPoolExecutor类的内部类,此处只讲最重要的构造函数和run方法
  1. private void addWorkerFailed(Worker w) {
  2.     final ReentrantLock mainLock = this.mainLock;
  3.     mainLock.lock();
  4.     try {
  5.         if (w != null)
  6.             workers.remove(w); // 线程启动失败时,需将前面添加的线程删除
  7.         decrementWorkerCount(); // ctl变量中的工作线程数-1
  8.         tryTerminate(); // 尝试将线程池转变成TERMINATE状态
  9.     } finally {
  10.         mainLock.unlock();
  11.     }
  12. }
复制代码
Worker实现了Runable接口,在调用start()方法候,实际执行的是run方法Worker实现了Runable接口,在调用start()方法候,实际执行的是run方法
  1. final void tryTerminate() {
  2.     for (;;) {
  3.         int c = ctl.get();
  4.         // 以下情况不会进入TERMINATED状态:
  5.         //(1)当前线程池为RUNNING状态
  6.         //(2)在TIDYING及以上状态
  7.         //(3)SHUTDOWN状态并且工作队列不为空
  8.         //(4)当前活跃线程数不等于0
  9.         if (isRunning(c) ||
  10.             runStateAtLeast(c, TIDYING) ||
  11.             (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
  12.             return;
  13.         if (workerCountOf(c) != 0) { // 工作线程数!=0
  14.             interruptIdleWorkers(ONLY_ONE); // 中断一个正在等待任务的线程
  15.             return;
  16.         }
  17.         final ReentrantLock mainLock = this.mainLock;
  18.         mainLock.lock();
  19.         try {
  20.             // 通过CAS自旋判断直到当前线程池运行状态为TIDYING并且活跃线程数为0
  21.             if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
  22.                 try {
  23.                     terminated(); // 调用线程terminated()
  24.                 } finally {
  25.                     ctl.set(ctlOf(TERMINATED, 0)); // 设置线程池状态为TERMINATED,工作线程数为0
  26.                     termination.signalAll(); // 通过调用Condition接口的signalAll()唤醒所有等待的线程
  27.                 }
  28.                 return;
  29.             }
  30.         } finally {
  31.             mainLock.unlock();
  32.         }
  33.         // else retry on failed CAS
  34.     }
  35. }
复制代码
从任务队列中取出一个任务
  1. private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
  2.     // 该worker正在运行的线程
  3.     final Thread thread;
  4.    
  5.     // 将要运行的初始任务
  6.     Runnable firstTask;
  7.    
  8.     // 每个线程的任务计数器
  9.     volatile long completedTasks;
  10.     // 构造方法   
  11.     Worker(Runnable firstTask) {
  12.         setState(-1); // 调用runWorker()前禁止中断
  13.         this.firstTask = firstTask;
  14.         this.thread = getThreadFactory().newThread(this); // 通过ThreadFactory创建一个线程
  15.     }
  16.     // 实现了Runnable接口的run方法
  17.     public void run() {
  18.         runWorker(this);
  19.     }
  20.    
  21.     ... // 此处省略了其他方法
  22. }
复制代码
总结一下哪些情况getTask()会返回null:

  • 线程池状态为SHUTDOWN且任务队列为空
  • 线程池状态为STOP、TIDYING、TERMINATED
  • 线程池线程数大于最大线程数
  • 线程可以被超时回收的情况下等待新任务超时
工作线程退出
  1. final void runWorker(Worker w) {
  2.     Thread wt = Thread.currentThread();
  3.     Runnable task = w.firstTask; // 获取工作线程中用来执行任务的线程实例
  4.     w.firstTask = null;
  5.     w.unlock(); // status设置为0,允许中断
  6.     boolean completedAbruptly = true; // 线程意外终止标志
  7.     try {
  8.         // 如果当前任务不为空,则直接执行;否则调用getTask()从任务队列中取出一个任务执行
  9.         while (task != null || (task = getTask()) != null) {
  10.             w.lock(); // 加锁,保证下方临界区代码的线程安全
  11.             // 如果状态值大于等于STOP且当前线程还没有被中断,则主动中断线程
  12.             if ((runStateAtLeast(ctl.get(), STOP) ||
  13.                  (Thread.interrupted() &&
  14.                   runStateAtLeast(ctl.get(), STOP))) &&
  15.                 !wt.isInterrupted())
  16.                 wt.interrupt(); // 中断当前线程
  17.             try {
  18.                 beforeExecute(wt, task); // 任务执行前的回调,空实现,可以在子类中自定义
  19.                 Throwable thrown = null;
  20.                 try {
  21.                     task.run(); // 执行线程的run方法
  22.                 } catch (RuntimeException x) {
  23.                     thrown = x; throw x;
  24.                 } catch (Error x) {
  25.                     thrown = x; throw x;
  26.                 } catch (Throwable x) {
  27.                     thrown = x; throw new Error(x);
  28.                 } finally {
  29.                     afterExecute(task, thrown); // 任务执行后的回调,空实现,可以在子类中自定义
  30.                 }
  31.             } finally {
  32.                 task = null; // 将循环变量task设置为null,表示已处理完成
  33.                 w.completedTasks++; // 当前已完成的任务数+1
  34.                 w.unlock();
  35.             }
  36.         }
  37.         completedAbruptly = false;
  38.     } finally {
  39.         processWorkerExit(w, completedAbruptly);
  40.     }
  41. }
复制代码
submit源码

提交任务到线程池有两种方法,一种是execute,另一种是submit。区别是execute没有返回值,submit是有返回值的,如果有异常抛出,submit同样可以获取异常结果。
  1. private Runnable getTask() {
  2.     boolean timedOut = false; // 通过timeOut变量表示线程是否空闲时间超时了
  3.     // 无限循环
  4.     for (;;) {
  5.         int c = ctl.get(); // 线程池信息
  6.         int rs = runStateOf(c); // 线程池当前状态
  7.         // 如果线程池状态>=SHUTDOWN并且工作队列为空 或 线程池状态>=STOP,则返回null,让当前worker被销毁
  8.         if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
  9.             decrementWorkerCount(); // 工作线程数-1
  10.             return null;
  11.         }
  12.         int wc = workerCountOf(c); // 获取当前线程池的工作线程数
  13.         // 当前线程是否允许超时销毁的标志
  14.         // 允许超时销毁:当线程池允许核心线程超时 或 工作线程数>核心线程数
  15.         boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
  16.         // 如果(当前线程数大于最大线程数 或 (允许超时销毁 且 当前发生了空闲时间超时))
  17.         // 且(当前线程数大于1 或 阻塞队列为空)
  18.         // 则减少worker计数并返回null
  19.         if ((wc > maximumPoolSize || (timed && timedOut))
  20.             && (wc > 1 || workQueue.isEmpty())) {
  21.             if (compareAndDecrementWorkerCount(c))
  22.                 return null;
  23.             continue;
  24.         }
  25.         try {
  26.             // 根据线程是否允许超时判断用poll还是take(会阻塞)方法从任务队列头部取出一个任务
  27.             Runnable r = timed ?
  28.                 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
  29.                 workQueue.take();//线程池重用逻辑:没有任务了就阻塞在这里,等待新的任务
  30.             if (r != null)
  31.                 return r; // 返回从队列中取出的任务
  32.             timedOut = true;
  33.         } catch (InterruptedException retry) {
  34.             timedOut = false;
  35.         }
  36.     }
  37. }
复制代码
submit中调用了newTaskFor方法来返回一个ftask对象,然后execute这个ftask对象,newTaskFor代码如下:
  1. private void processWorkerExit(Worker w, boolean completedAbruptly) {
  2.     // 如果completedAbruptly为true则表示任务执行过程中抛出了未处理的异常
  3.     // 所以还没有正确地减少worker计数,这里需要减少一次worker计数
  4.     if (completedAbruptly)
  5.         decrementWorkerCount();
  6.     final ReentrantLock mainLock = this.mainLock;
  7.     mainLock.lock();
  8.     try {
  9.         // 把将被销毁的线程已完成的任务数累加到线程池的完成任务总数上
  10.         completedTaskCount += w.completedTasks;
  11.         workers.remove(w); // 从工作线程集合中移除该工作线程
  12.     } finally {
  13.         mainLock.unlock();
  14.     }
  15.     // 尝试结束线程池
  16.     tryTerminate();
  17.     int c = ctl.get();
  18.     // 如果是RUNNING 或 SHUTDOWN状态
  19.     if (runStateLessThan(c, STOP)) {
  20.         // worker是正常执行完
  21.         if (!completedAbruptly) {
  22.             // 如果允许核心线程超时则最小线程数是0,否则最小线程数等于核心线程数
  23.             int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
  24.             // 如果阻塞队列非空,则至少要有一个线程继续执行剩下的任务
  25.             if (min == 0 && ! workQueue.isEmpty())
  26.                 min = 1;
  27.             // 如果当前线程数已经满足最小线程数要求,则不需要再创建替代线程
  28.             if (workerCountOf(c) >= min)
  29.                 return; // replacement not needed
  30.         }
  31.         // 重新创建一个worker来代替被销毁的线程
  32.         addWorker(null, false);
  33.     }
  34. }
复制代码
newTaskFor又调用FutureTask的有参构造器来创建一个futureTask实例,代码如下:
  1. // AbstractExecutorService.submit
  2. public Future<?> submit(Runnable task) {
  3.     if (task == null) throw new NullPointerException();
  4.     RunnableFuture<Void> ftask = newTaskFor(task, null);
  5.     execute(ftask);
  6.     return ftask;
  7. }
复制代码
这个有参构造器中又调用了Executors的静态方法callable创建一个callable实例来赋值给futureTask的callable属性,代码如下:
  1. // AbstractExecutorService.newTaskFor
  2. protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
  3.     return new FutureTask<T>(runnable, value);
  4. }
复制代码
最后还是使用了RunnableAdapter来包装这个task,代码如下:
  1. // FutureTask有参构造器
  2. public FutureTask(Runnable runnable, V result) {
  3.     this.callable = Executors.callable(runnable, result);
  4.     this.state = NEW;       // ensure visibility of callable
  5. }
复制代码
梳理一下整个流程,run和call的关系的伪代码如下
  1. // Executors.callable
  2. public static <T> Callable<T> callable(Runnable task, T result) {
  3.     if (task == null)
  4.         throw new NullPointerException();
  5.     return new RunnableAdapter<T>(task, result);
  6. }
复制代码
为什么要这么麻烦封装一层又一层呢?
可能是为了适配。submit的返回值是futureTask,但是传给submit的是个runnable,然后submit会把这个runnable继续传给futureTask,futureTask的结果值是null,但是又由于futureTask的run方法已经被重写成执行call方法了,所以只能在call方法里面跑真正的run方法了

来源:程序园用户自行投稿发布,如果侵权,请联系站长删除
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!
您需要登录后才可以回帖 登录 | 立即注册