Executor Framework Java

2025年5月3日 | 阅读 6 分钟

管理工作线程的组件集合，用于高效管理工作线程，被称为 Executor Framework。Executor API 通过 Executors 将任务的执行与实际要执行的任务分离开来。Executor framework 是 生产者-消费者 模式的实现。java.util.concurrent.Executors 类提供了一组用于创建工作线程 ThreadPool 的方法。

要使用 Executor framework，我们必须创建一个线程池来执行任务，并将该任务提交到该线程池。

现在，我们脑海中浮现的问题是，为什么我们需要创建这样的线程池，当我们已经有了 java.lang.Thread 类来创建对象，以及 Runnable/Callable 接口来实现并行性时？那么，创建这样的线程池的原因如下：

我们需要创建大量线程，以便在没有任何节流的情况下为每个进程添加新线程。这会导致需要更多内存并浪费资源。当每个线程被交换时，CPU 会花费过多的时间。
当为执行新任务创建新线程时，会导致线程创建的开销。为了管理这个线程的生命周期，执行时间也会相应增加。

Executor 的类型

在 Java 中，有不同类型的 executors 可用，如下所示：

1) SingleThreadExecutor

SingleThreadExecutor 是一种特殊的 Executor，它只有一个线程。当我们需要按顺序执行任务时使用它。如果某个线程在执行任务时因某种错误或异常而死亡，则会创建一个新线程，并且所有后续任务都将在该新线程中执行。

2) FixedThreadPool(n)

FixedThreadPool 是另一种特殊的 Executor，它是一个具有固定数量线程的线程池。通过这个 Executor，已提交的任务由 n 个线程执行。如果我们要在提交前一个任务后执行更多任务，它们会存储在 LinkedBlockingQueue 中，直到前一个任务完成。n 表示底层处理器支持的线程总数。

3) CachedThreadPool

CachedThreadPool 是一种特殊的线程池，用于执行短暂的并行任务。缓存线程池没有固定数量的线程。当一个新任务到来，而此时所有线程都忙于执行其他任务时，池会创建一个新线程并将其添加到 Executor 中。当一个线程空闲时，它会执行新任务。当线程空闲六十秒后，线程会被终止并从缓存中移除。

4) ScheduledExecutor

ScheduledExecutor 是我们用于定期运行某个任务的另一种特殊 Executor。当我们需要延迟某个任务时也使用它。

ScheduledExecutorService scheduledExecService = Executors.newScheduledThreadPool(1);

scheduleAtFixedRate 和 scheduleWithFixedDelay 是在 ScheduledExecutor 中用于调度任务的两个方法。scheduleAtFixedRate 方法在上一任务结束后以固定间隔执行任务。scheduleWithFixedDelay 方法在当前任务完成后开始倒计时。这两个方法之间的主要区别在于它们对计划作业连续执行之间延迟的解释。这两个方法都以以下方式使用：

scheduledExecService.scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)

scheduledExecService.scheduleWithFixedDelay(Runnable command, long initialDelay, long period, TimeUnit unit)

让我们举一个例子来理解 Executor 实际是如何用于执行任务的。我们将创建一个任务，并尝试分别通过简单的 Executor 和 ThreadPoolExecutor 来执行它。

Task1.java

//creating Task1 class that implements the Runnable interface
public class Task1 implements Runnable {
	
	//initialize threadNo variable 
    private String threadNo;
    
	//using constructor to set value to the threadNo variable
    public Task1(String no){
        this.threadNo = no;
    }
	
	//using run() method by overriding it
    @Override
    public void run() {
		//printing thread information 
        System.out.println(Thread.currentThread().getName()+" start execution. Thread No = "+threadNo);
		
		//calling processThread() method for processing thread
        processThread();
		
		//printing a statement to show the end of the thread
        System.out.println(Thread.currentThread().getName()+" stop execution.");
    }
	
	//creating processThread() method to execute thread
    private void processThread() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
	
	//overriding toString() method 
    @Override
    public String toString(){
        return this.threadNo;
    }
}

SimpleExecutor.java

//importing classes 
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

//creating SimpleExecutor class
public class SimpleExecutor {
	//main() method start
    public static void main(String[] args) {
        //creating an object of ExecutorService with fixed thread pool 5
		ExecutorService executorService = Executors.newFixedThreadPool(5);
        for (int j = 0; j < 5; j++) {
            //creating instance of the Task1 class and pass a value to its constructor 
			Runnable task = new Task1("" + j);
			
			//executing task using execute() method of the executor
            executorService.execute(task);
          }
		//closing executor
        executorService.shutdown();
        
		while (!executorService.isTerminated()) {
        }
		System.out.println("Finished all threads");
    }
}

输出

让我们以 ThreadPoolExecutor 为例来理解它与 SimpleExecutor 有何不同。

RejectedExecutionHandlerDemo.java

RejectedExecutionHandlerDemo 类用于处理从工作队列中被拒绝的任务。当线程池大小达到限制时，RejectedExecutionHandler 会处理那些无法放入工作线程的任务。我们将按以下方式实现 RejectedExecutionHandler.java 类：

//importing RejectedExecutionHandler, and ThreadPoolExecutor classes 
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

//creating RejectedExecutionHandlerDemo class to handle rejected execution
public class RejectedExecutionHandlerDemo implements RejectedExecutionHandler {

    //defining abstract method of RejectedExecutionHandler class
	@Override
    public void rejectedExecution(Runnable runnableObject, ThreadPoolExecutor executorObjet) {
        //printing the rejected thread
		System.out.println(runnableObject.toString() + " is rejected");
    }

}

NewTask.java

我们创建 NewTask.java 类来创建一个线程，该线程监控 Executor 并在特定时间间隔打印其信息。ThreadPoolExecutor 类中有几种方法可用于获取 Executor 的当前状态、活动线程数、任务数和池大小。我们按以下方式创建监视器线程：

//importing ThreadPoolExecutor class
import java.util.concurrent.ThreadPoolExecutor;

//creating NewTask class that implements the Runnable interface
public class NewTask implements Runnable
{
    private ThreadPoolExecutor exe;//creating a private variable of type ThreadPoolExecutor
    private int delay; //creating a private variable of type integer for delay
    private boolean run = true; //creating a private boolean variable to determine whether 
	
	//using constructor to set the value to the exe and delay variable
    public NewTask(ThreadPoolExecutor exe, int delay)
    {
		this.exe = exe;
        this.delay = delay;
    }
	
	//creating shutdown() method to update the value of the boolean variable run 
    public void shutdown(){
        this.run = false;
    }
	
	//overriding the run() method of the Runnable interface
    @Override
    public void run()
    {
		//running task
        while(run){
                System.out.println(
                    String.format("[monitor [%d/%d] Number of active threads = %d, Number of completed tasks = %d, Number of tasks = %d, Shutdown = %s, Terminated = %s",
                        this.exe.getPoolSize(),
                        this.exe.getCorePoolSize(),
                        this.exe.getActiveCount(),
                        this.exe.getCompletedTaskCount(),
                        this.exe.getTaskCount(),
                        this.exe.isShutdown(),
                        this.exe.isTerminated()));
                try {
                    Thread.sleep(this.delay*1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
        }
            
    }
}

ThreadPoolExecutorExample.java

它用于使用 ThreadPoolExecutor 创建线程池。

//importing classes required for implementing ThreadPoolExecutor
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

//creating ThreadPoolExecutorExample to execute task
public class ThreadPoolExecutorExample {
	
	//main method start
    public static void main(String args[]) throws InterruptedException{
        
		//creating an instance of the RejectedExecutionHandlerExample class
        RejectedExecutionHandlerDemo rejectionHandlerObj = new RejectedExecutionHandlerDemo();
        
		//creating an instance of the ThreadFactory class
        ThreadFactory t1 = Executors.defaultThreadFactory();
        
		//creating an instance of the ThreadPoolExecutor class
        ThreadPoolExecutor exePool = new ThreadPoolExecutor(2, 4, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2), t1, rejectionHandlerObj);
        
		//starting the NewTask thread
		NewTask task = new NewTask(exePool, 3);
        Thread th = new Thread(task);
        th.start();
        //submit job to the pool
        for(int j=0; j<10; j++){
            exePool.execute(new Task1("cmd"+j));
        }
        
        Thread.sleep(30000);
        //shutting down the job pool
        exePool.shutdown();
		
        //shutting down the th thread
        Thread.sleep(5000);
        task.shutdown();
        
    }
}

输出

请注意活动线程数、已完成任务数和任务数的变化。shutdown() 方法用于完成所有已提交任务的执行并终止线程池。

下一主题数组中缺失的问题-java

Executor Framework Java

Executor 的类型

1) SingleThreadExecutor

2) FixedThreadPool(n)

3) CachedThreadPool

4) ScheduledExecutor

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Java Conversion

Java Misc

Executor Framework Java

Executor 的类型

1) SingleThreadExecutor

2) FixedThreadPool(n)

3) CachedThreadPool

4) ScheduledExecutor

相关帖子

Java 中的 getSource() 方法

Java 中的三个字符串的 LCS

Java 中的 Kynea 数

Java 中的 replace() vs replaceAll()

Java 中的结构化并发

Unsigned Right Shift Operator in Java

Java 中不使用额外空间合并两个排序数组

Java 中的 BFS 算法

Java 泛型的优点

Java 中的 Pangram 程序

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器