`

Executor/Executors

阅读更多

1,Executor

 

public interface Executor

执行已提交的 Runnable 任务的对象。此接口提供一种将任务提交与每个任务将如何运行的机制(包括线程使用的细节、调度等)分离开来的方法。通常使用 Executor 而不是显式地创建线程。例如,可能会使用以下方法,而不是为一组任务中的每个任务调用 new Thread(new(RunnableTask())).start()

 Executor executor = anExecutor;
 executor.execute(new RunnableTask1());
 executor.execute(new RunnableTask2());
 ...
 

不过,Executor 接口并没有严格地要求执行是异步的。在最简单的情况下,执行程序可以在调用者的线程中立即运行已提交的任务:

 class DirectExecutor implements Executor {
     public void execute(Runnable r) {
         r.run();
     }
 }

更常见的是,任务是在某个不是调用者线程的线程中执行的。以下执行程序将为每个任务生成一个新线程。

 class ThreadPerTaskExecutor implements Executor {
     public void execute(Runnable r) {
         new Thread(r).start();
     }
 }

许多 Executor 实现都对调度任务的方式和时间强加了某种限制。以下执行程序使任务提交与第二个执行程序保持连续,这说明了一个复合执行程序。

 class SerialExecutor implements Executor {
     final Queue<Runnable> tasks = new ArrayDeque<Runnable>();
     final Executor executor;
     Runnable active;

SerialExecutor(Executor executor) {
         this.executor = executor;
     }

public synchronized void execute(final Runnable r) {
         tasks.offer(new Runnable() {
             public void run() {
                 try {
                     r.run();
                 } finally {
                     scheduleNext();
                 }
             }
         });
         if (active == null) {
             scheduleNext();
         }
     }

protected synchronized void scheduleNext() {
         if ((active = tasks.poll()) != null) {
             executor.execute(active);
         }
     }
 }

此包中提供的 Executor 实现实现了 ExecutorService,这是一个使用更广泛的接口。ThreadPoolExecutor 类提供一个可扩展的线程池实现。Executors 类为这些 Executor 提供了便捷的工厂方法。

内存一致性效果:线程中将 Runnable 对象提交到 Executor 之前的操作 happen-before 其执行开始(可能在另一个线程中)。

 

execute

void execute(Runnable command)

在未来某个时间执行给定的命令。该命令可能在新的线程、已入池的线程或者正调用的线程中执行,这由 Executor 实现决定。

参数:

command - 可运行的任务

抛出:

RejectedExecutionException - 如果不能接受执行此任务。

NullPointerException - 如果命令为 null

 

2,ExecutorService

 

 

public interface ExecutorService

extends Executor

Executor 提供了管理终止的方法,以及可为跟踪一个或多个异步任务执行状况而生成 Future 的方法。

可以关闭 ExecutorService,这将导致其拒绝新任务。提供两个方法来关闭 ExecutorServiceshutdown() 方法在终止前允许执行以前提交的任务,而 shutdownNow() 方法阻止等待任务启动并试图停止当前正在执行的任务。在终止时,执行程序没有任务在执行,也没有任务在等待执行,并且无法提交新任务。应该关闭未使用的 ExecutorService 以允许回收其资源。

通过创建并返回一个可用于取消执行和/或等待完成的 Future,方法 submit 扩展了基本方法 Executor.execute(java.lang.Runnable)。方法 invokeAny  invokeAll 是批量执行的最常用形式,它们执行任务 collection,然后等待至少一个,或全部任务完成(可使用 ExecutorCompletionService 类来编写这些方法的自定义变体)。

Executors 类提供了用于此包中所提供的执行程序服务的工厂方法。

用法示例

下面给出了一个网络服务的简单结构,这里线程池中的线程作为传入的请求。它使用了预先配置的 Executors.newFixedThreadPool(int) 工厂方法:

 class NetworkService implements Runnable {
    private final ServerSocket serverSocket;
    private final ExecutorService pool;

public NetworkService(int port, int poolSize)
        throws IOException {
      serverSocket = new ServerSocket(port);
      pool = Executors.newFixedThreadPool(poolSize);
    }
 
    public void run() { // run the service
      try {
        for (;;) {
          pool.execute(new Handler(serverSocket.accept()));
        }
      } catch (IOException ex) {
        pool.shutdown();
      }
    }
  }

class Handler implements Runnable {
    private final Socket socket;
    Handler(Socket socket) { this.socket = socket; }
    public void run() {
      // read and service request on socket
    }
 }
 

下列方法分两个阶段关闭 ExecutorService。第一阶段调用 shutdown 拒绝传入任务,然后调用 shutdownNow(如有必要)取消所有遗留的任务:

 void shutdownAndAwaitTermination(ExecutorService pool) {
   pool.shutdown(); // Disable new tasks from being submitted
   try {
     // Wait a while for existing tasks to terminate
     if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
       pool.shutdownNow(); // Cancel currently executing tasks
       // Wait a while for tasks to respond to being cancelled
       if (!pool.awaitTermination(60, TimeUnit.SECONDS))
           System.err.println("Pool did not terminate");
     }
   } catch (InterruptedException ie) {
     // (Re-)Cancel if current thread also interrupted
     pool.shutdownNow();
     // Preserve interrupt status
     Thread.currentThread().interrupt();
   }
 }
 

内存一致性效果:线程中向 ExecutorService 提交 Runnable  Callable 任务之前的操作 happen-before 由该任务所提取的所有操作,后者依次 happen-before 通过 Future.get() 获取的结果。

 

3,AbstractExecutorService

 

public abstract class AbstractExecutorService

extends Object

implements ExecutorService

提供 ExecutorService 执行方法的默认实现。此类使用 newTaskFor 返回的 RunnableFuture 实现 submit、invokeAny  invokeAll 方法,默认情况下,RunnableFuture 是此包中提供的 FutureTask 类。例如,submit(Runnable) 的实现创建了一个关联 RunnableFuture 类,该类将被执行并返回。子类可以重写 newTaskFor 方法,以返回 FutureTask 之外的 RunnableFuture 实现。

扩展示例。以下是一个类的简要介绍,该类定制 ThreadPoolExecutor 使用 CustomTask 类替代默认 FutureTask

 public class CustomThreadPoolExecutor extends ThreadPoolExecutor {

static class CustomTask<V> implements RunnableFuture<V> {...}

protected <V> RunnableFuture<V> newTaskFor(Callable<V> c) {
       return new CustomTask<V>(c);
   }
   protected <V> RunnableFuture<V> newTaskFor(Runnable r, V v) {
       return new CustomTask<V>(r, v);
   }
   // ... add constructors, etc.
 }
 

4,ThreadPoolExecutor

 

public class ThreadPoolExecutor

extends AbstractExecutorService

一个 ExecutorService,它使用可能的几个池线程之一执行每个提交的任务,通常使用 Executors 工厂方法配置。

线程池可以解决两个不同问题:由于减少了每个任务调用的开销,它们通常可以在执行大量异步任务时提供增强的性能并且还可以提供绑定和管理资源(包括执行任务集时使用的线程)的方法。每个 ThreadPoolExecutor 还维护着一些基本的统计数据,如完成的任务数。

为了便于跨大量上下文使用,此类提供了很多可调整的参数和扩展钩子 (hook)。但是,强烈建议程序员使用较为方便的 Executors 工厂方法 Executors.newCachedThreadPool()(无界线程池,可以进行自动线程回收)、Executors.newFixedThreadPool(int)(固定大小线程池)和 Executors.newSingleThreadExecutor()(单个后台线程),它们均为大多数使用场景预定义了设置。否则,在手动配置和调整此类时,使用以下指导:

核心和最大池大小

ThreadPoolExecutor 将根据 corePoolSize(参见 getCorePoolSize())和 maximumPoolSize(参见 getMaximumPoolSize())设置的边界自动调整池大小。当新任务在方法 execute(java.lang.Runnable) 中提交时,如果运行的线程少于 corePoolSize,则创建新线程来处理请求,即使其他辅助线程是空闲的。如果运行的线程多于 corePoolSize 而少于 maximumPoolSize,则仅当队列满时才创建新线程。如果设置的 corePoolSize maximumPoolSize 相同,则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值(如 Integer.MAX_VALUE),则允许池适应任意数量的并发任务。在大多数情况下,核心和最大池大小仅基于构造来设置,不过也可以使用 setCorePoolSize(int)  setMaximumPoolSize(int) 进行动态更改。

按需构造

默认情况下,即使核心线程最初只是在新任务到达时才创建和启动的,也可以使用方法 prestartCoreThread()  prestartAllCoreThreads() 对其进行动态重写。如果构造带有非空队列的池,则可能希望预先启动线程。

创建新线程

使用 ThreadFactory 创建新线程。如果没有另外说明,则在同一个 ThreadGroup 中一律使用 Executors.defaultThreadFactory() 创建线程,并且这些线程具有相同的 NORM_PRIORITY 优先级和非守护进程状态。通过提供不同的 ThreadFactory,可以改变线程的名称、线程组、优先级、守护进程状态,等等。如果从 newThread 返回 null  ThreadFactory 未能创建线程,则执行程序将继续运行,但不能执行任何任务。

保持活动时间

如果池中当前有多于 corePoolSize 的线程,则这些多出的线程在空闲时间超过 keepAliveTime 时将会终止(参见 getKeepAliveTime(java.util.concurrent.TimeUnit))。这提供了当池处于非活动状态时减少资源消耗的方法。如果池后来变得更为活动,则可以创建新的线程。也可以使用方法 setKeepAliveTime(long, java.util.concurrent.TimeUnit) 动态地更改此参数。使用 Long.MAX_VALUE TimeUnit.NANOSECONDS 的值在关闭前有效地从以前的终止状态禁用空闲线程。默认情况下,保持活动策略只在有多于 corePoolSizeThreads 的线程时应用。但是只要 keepAliveTime 值非 0allowCoreThreadTimeOut(boolean) 方法也可将此超时策略应用于核心线程。

排队

所有 BlockingQueue 都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互:

  • 如果运行的线程少于 corePoolSize,则 Executor 始终首选添加新的线程,而不进行排队。
  • 如果运行的线程等于或多于 corePoolSize,则 Executor 始终首选将请求加入队列,而不添加新的线程。
  • 如果无法将请求加入队列,则创建新的线程,除非创建此线程超出 maximumPoolSize,在这种情况下,任务将被拒绝。

排队有三种通用策略:

  1. 直接提交。工作队列的默认选项是 SynchronousQueue,它将任务直接提交给线程而不保持它们。在此,如果不存在可用于立即运行任务的线程,则试图把任务加入队列将失败,因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时,此策略允许无界线程具有增长的可能性。

newCachedThreadPool

  1. 无界队列。使用无界队列(例如,不具有预定义容量的 LinkedBlockingQueue)将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。这样,创建的线程就不会超过 corePoolSize。(因此,maximumPoolSize 的值也就无效了。)当每个任务完全独立于其他任务,即任务执行互不影响时,适合于使用无界队列;例如,在 Web 页服务器中。这种排队可用于处理瞬态突发请求,当命令以超过队列所能处理的平均数连续到达时,此策略允许无界线程具有增长的可能性。

newFixedThreadPool/newSingleThreadExecutor

  1. 有界队列。当使用有限的 maximumPoolSizes 时,有界队列(如 ArrayBlockingQueue)有助于防止资源耗尽,但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷:使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销,但是可能导致人工降低吞吐量。如果任务频繁阻塞(例如,如果它们是 I/O 边界),则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小,CPU 使用率较高,但是可能遇到不可接受的调度开销,这样也会降低吞吐量。

被拒绝的任务

当 Executor 已经关闭,并且 Executor 将有限边界用于最大线程和工作队列容量,且已经饱和时,在方法 execute(java.lang.Runnable) 中提交的新任务将被拒绝。在以上两种情况下,execute 方法都将调用其 RejectedExecutionHandler  RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) 方法。下面提供了四种预定义的处理程序策略:

  1. 在默认的 ThreadPoolExecutor.AbortPolicy 中,处理程序遭到拒绝将抛出运行时 RejectedExecutionException
  2.  ThreadPoolExecutor.CallerRunsPolicy 中,线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制,能够减缓新任务的提交速度。
  3.  ThreadPoolExecutor.DiscardPolicy 中,不能执行的任务将被删除。
  4.  ThreadPoolExecutor.DiscardOldestPolicy 中,如果执行程序尚未关闭,则位于工作队列头部的任务将被删除,然后重试执行程序(如果再次失败,则重复此过程)。

定义和使用其他种类的 RejectedExecutionHandler 类也是可能的,但这样做需要非常小心,尤其是当策略仅用于特定容量或排队策略时。

钩子 (hook) 方法

此类提供 protected 可重写的 beforeExecute(java.lang.Thread, java.lang.Runnable)  afterExecute(java.lang.Runnable, java.lang.Throwable) 方法,这两种方法分别在执行每个任务之前和之后调用。它们可用于操纵执行环境;例如,重新初始化 ThreadLocal、搜集统计信息或添加日志条目。此外,还可以重写方法 terminated() 来执行 Executor 完全终止后需要完成的所有特殊处理。

如果钩子 (hook) 或回调方法抛出异常,则内部辅助线程将依次失败并突然终止。

队列维护

方法 getQueue() 允许出于监控和调试目的而访问工作队列。强烈反对出于其他任何目的而使用此方法。remove(java.lang.Runnable)  purge() 这两种方法可用于在取消大量已排队任务时帮助进行存储回收。

终止

程序 AND 不再引用的池没有剩余线程会自动 shutdown。如果希望确保回收取消引用的池(即使用户忘记调用 shutdown()),则必须安排未使用的线程最终终止:设置适当保持活动时间,使用 0 核心线程的下边界和/或设置 allowCoreThreadTimeOut(boolean)

扩展示例。此类的大多数扩展可以重写一个或多个受保护的钩子 (hook) 方法。例如,下面是一个添加了简单的暂停/恢复功能的子类:

 class PausableThreadPoolExecutor extends ThreadPoolExecutor {
   private boolean isPaused;
   private ReentrantLock pauseLock = new ReentrantLock();
   private Condition unpaused = pauseLock.newCondition();

public PausableThreadPoolExecutor(...) { super(...); }
 
   protected void beforeExecute(Thread t, Runnable r) {
     super.beforeExecute(t, r);
     pauseLock.lock();
     try {
       while (isPaused) unpaused.await();
     } catch(InterruptedException ie) {
       t.interrupt();
     } finally {
       pauseLock.unlock();
     }
   }
 
   public void pause() {
     pauseLock.lock();
     try {
       isPaused = true;
     } finally {
       pauseLock.unlock();
     }
   }
 
   public void resume() {
     pauseLock.lock();
     try {
       isPaused = false;
       unpaused.signalAll();
     } finally {
       pauseLock.unlock();
     }
   }
 }

 

5,ScheduledThreadPoolExecutor

 

public class ScheduledThreadPoolExecutor

extends ThreadPoolExecutor

implements ScheduledExecutorService

ThreadPoolExecutor它可另行安排在给定的延迟后运行命令,或者定期执行命令。需要多个辅助线程时,或者要求 ThreadPoolExecutor 具有额外的灵活性或功能时,此类要优于 Timer

一旦启用已延迟的任务就执行它,但是有关何时启用,启用后何时执行则没有任何实时保证。按照提交的先进先出 (FIFO) 顺序来启用那些被安排在同一执行时间的任务。

虽然此类继承自 ThreadPoolExecutor,但是几个继承的调整方法对此类并无作用。特别是,因为它作为一个使用 corePoolSize 线程和一个无界队列的固定大小的池,所以调整 maximumPoolSize 没有什么效果。

扩展注意事项:此类重写 AbstractExecutorService  submit 方法,以生成内部对象控制每个任务的延迟和调度。若要保留功能性,子类中任何进一步重写的这些方法都必须调用超类版本,超类版本有效地禁用附加任务的定制。但是,此类提供替代受保护的扩展方法 decorateTask(为 Runnable  Callable 各提供一种版本),可定制用于通过 execute、submit、schedule、scheduleAtFixedRate  scheduleWithFixedDelay 进入的执行命令的具体任务类型。默认情况下,ScheduledThreadPoolExecutor 使用一个扩展 FutureTask 的任务类型。但是,可以使用下列形式的子类修改或替换该类型。

 public class CustomScheduledExecutor extends ScheduledThreadPoolExecutor {

static class CustomTask<V> implements RunnableScheduledFuture<V> { ... }

protected <V> RunnableScheduledFuture<V> decorateTask(
                Runnable r, RunnableScheduledFuture<V> task) {
       return new CustomTask<V>(r, task);
   }

protected <V> RunnableScheduledFuture<V> decorateTask(
                Callable<V> c, RunnableScheduledFuture<V> task) {
       return new CustomTask<V>(c, task);
   }
   // ... add constructors, etc.
 }

 

6,Executors

 

public class Executors

extends Object

此包中所定义的 ExecutorExecutorServiceScheduledExecutorServiceThreadFactory  Callable 类的工厂和实用方法。此类支持以下各种方法:

 

  • 创建并返回设置有常用配置字符串的 ExecutorService 的方法。
  • 创建并返回设置有常用配置字符串的 ScheduledExecutorService 的方法。
  • 创建并返回“包装的”ExecutorService 方法,它通过使特定于实现的方法不可访问来禁用重新配置。
  • 创建并返回 ThreadFactory 的方法,它可将新创建的线程设置为已知的状态。
  • 创建并返回非闭包形式的 Callable 的方法,这样可将其用于需要 Callable 的执行方法中。
分享到:
评论

相关推荐

    Executor,Executors,ExecutorService比较.docx

    【Executor、Executors和ExecutorService详解】 在Java并发编程中,`Executor`、`Executors`和`ExecutorService`是核心组件,它们帮助开发者高效管理线程资源,提高程序的并发性能。理解这三个概念的区别和用途是...

    java并发编程:Executor、Executors、ExecutorService.docx

    Java并发编程中的Executor、Executors和ExecutorService是Java并发编程框架的重要组成部分,它们为开发者提供了高效管理和控制线程执行的工具。以下是对这些概念的详细解释: 1. Executor: Executor是一个接口,它...

    java 中Executor, ExecutorService 和 Executors 间的不同

    Java 中 Executor, ExecutorService 和 Executors 的不同 Java 中的 Executor, ExecutorService 和 Executors 是 Java Executor 框架的重要组件,用于提供线程池的功能。在 Java 1.5 之后,Executor 框架提供了多种...

    thrift服务端和客户端实现Nifty.zip

    ExecutorService bossExecutor = Executors.newCachedThreadPool(); ExecutorService workerExecutor = Executors.newCachedThreadPool(); // Start the server server.start(bossExecutor, workerExecutor); ...

    顶层接口Executors详解

    顶层接口Executors详解 Executors框架是Java语言中用于异步执行任务的高级接口,旨在提供一个高效、灵活、可扩展的任务执行机制。该框架提供了一个两级调度模型,第一级是用户级的调度器,第二级是操作系统内核的...

    几种方法编写线程代码

    ExecutorService executor = Executors.newFixedThreadPool(5); executor.execute(new Runnable() { public void run() { // 任务代码 } }); // 当不再提交任务时,关闭ExecutorService executor.shutdown(); ```...

    Java多线程实现的四种方式

    ExecutorService executor = Executors.newFixedThreadPool(5); // 创建固定大小的线程池 executor.execute(new Runnable() { @Override public void run() { // 多线程代码 } }); executor.shutdown(); // 关闭...

    掌握并发的钥匙:Java Executor框架深度解析

    ExecutorService executor = Executors.newCachedThreadPool(); ``` #### 代码示例:创建自定义线程池 ```java int corePoolSize = 5; int maximumPoolSize = 10; long keepAliveTime = 1L; TimeUnit unit = ...

    java在服务启动的时候启动定时器

    ScheduledExecutorService executor = Executors.newScheduledThreadPool(1); Runnable task = () -&gt; System.out.println("定时任务执行"); executor.scheduleAtFixedRate(task, 0, 5, TimeUnit.SECONDS); // 当...

    Java计时器

    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() -&gt; { System.out.println("Task executed at " + new Date()); }, 0, 2, TimeUnit....

    Java线程创建的四种方式

    ExecutorService executor = Executors.newFixedThreadPool(5); for (int i = 0; i ; i++) { executor.execute(new Runnable() { public void run() { // 线程执行的任务 } }); } executor.shutdown(); // ...

    1_Executor源码阅读1

    Java中的`Executor`接口是并发编程的核心组件,它位于`java.util.concurrent`包下,主要目的是接收客户端提交的任务(以`Runnable`实例的形式)并负责执行。`Executor`接口简化了多线程编程,使得任务的创建和执行...

    ThreadPoolExecutor的使用和Android常见的4种线程池使用介绍

    ThreadPoolExecutor的使用和Android常见的4种线程池使用介绍

    Azkaban报错-azkaban.executor.ExecutorManagerException: No active executors found

    主要介绍了Azkaban报错-azkaban.executor.ExecutorManagerException: No active executors found,本文给大家介绍的非常详细,需要的朋友可以参考下

    使用Java匿名内部类实现一个简单的线程池.txt

    ExecutorService executor = Executors.newFixedThreadPool(5); // 循环提交任务 for (int i = 0; i ; i++) { Runnable task = new Runnable() { @Override public void run() { System.out.println("Task...

    JAVA面试题 用于面试前练手,有需要的自取

    本文将详细讲解 Java 面试题,包括线程与进程的区别、多线程中的上下文切换、死锁与活锁的区别、Java 中守护线程和本地线程的区别、为什么使用 Executor 框架、Java 中 Executor 和 Executors 的区别、Java 中用到的...

    Java多线程示例 可以直接运行

    ExecutorService executor = Executors.newSingleThreadExecutor(); Future&lt;Integer&gt; future = executor.submit(new Callable() { @Override public Integer call() throws Exception { return someCalculation()...

    Java-Executor并发框架.docx

    Executors类是一个工厂类,提供了一系列静态方法,用于创建不同类型的ExecutorService实例,满足不同场景的需求: 1. `newSingleThreadExecutor()`: 创建一个单线程的ExecutorService。所有任务都会按顺序串行执行...

    Java多线程 例题

    ExecutorService executor = Executors.newSingleThreadExecutor(); Future&lt;Integer&gt; future = executor.submit(() -&gt; calculate()); Integer result = future.get(); // 获取结果,会阻塞直到计算完成 executor....

    Java多线程之Executor框架.docx

    `Executors`是`ExecutorService`的工厂类,提供了一些静态方法来创建不同类型的线程池,如固定大小的线程池、单线程的线程池、缓存线程池等。这些线程池可以根据需求实现更高效、更可控的并发处理。 在实际应用中,...

Global site tag (gtag.js) - Google Analytics