Java的线程既是工作单元,也是执行机制。JDK5开始,把工作单元和执行机制分离开,工作单元包括Runnable
和Callable
,而执行机制由Executor
框架提供。
Executor框架简介
Executor框架的结构和成员
-
Executor框架的结构:主要由下面3部分组成。
- 任务:被执行的任务需要实现Runnable接口或Callable接口。
- 任务的执行:包括任务执行机制的核心接口Executor,以及继承自Executor的ExecutorService接口。Executor框架有两个关键类实现了ExecutorService接口(
ThreadPoolExecutor
和ScheduledThreadPoolExecutor
)。 - 异步计算的结果:包括接口Future和实现Future接口的FutureTask类。
-
主要的类和接口
-
Executor
是一个接口,是Executor框架的基础,它将任务的提交和任务的执行分离开来。 -
ThreadPoolExecutor
是线程池的核心实现类,用来执行被提交的任务。 -
ScheduledThreadPoolExecutor
是一个实现类,可以在给定的延迟后运行命令,或者定期执行命令。ScheduledThreadPoolExecutor
比Timer
更灵活,功能更强大。 -
Future
接口和实现Future
接口的FutureTask
类,代表异步计算的结果。 -
Runnable
和Callable
接口的实现类,具体执行任务的类。
-
-
Executor框架的成员
介绍Executor框架的主要成员:ThreadPoolExecutor
、ScheduledThreadPoolExecutor
、Future
接口、Runnable
接口、Callable
接口、Executors
。-
ThreadPoolExecutor
通常使用工厂类Executors来创建。有3中类型。-
FixedThreadPool
:创建使用固定线程数的。适用于需要限制当前线程数量的应用场景。
-
-
public static ExecutorService newFixedThreadPool(int nThreads)
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory)
* `SingleThreadExecutor`:创建单个线程。
public static ExecutorService newSingleThreadExecutor()
* `CachedThreadPool`:大小无界的线程池,适用于执行很多的短期异步任务的小程序,或者是负载较轻的服务器。
corePoolSize
为0,maximumPoolSize
为Integer.MAX_VALUE
,工作队列为SynchronousQueue
。会出现极端情况,即用户提交任务速度高于线程处理任务速度,会不断的创建新线程,直到耗尽CPU和内存。每个线程在结束任务60s之后,会被终止。
-
ScheduledThreadPoolExecutor
通过工厂类Executor来创建,有2种类型。-
ScheduledThreadPoolExecutor
:包含若干个线程的ScheduledThreadPoolExecutor
。 -
SingleThreadScheduledExecutor
:只包含一个线程的ScheduledThreadPoolExecutor
。
-
Future接口
使用submit提交的时候,会返回一个实现Future接口的对象。Runnable接口和Callable接口
都可以被ThreadPoolExecutor
执行,区别是Runnable接口不会返回结果,而Callable接口可以返回结果。
可以把一个Runnable对象包装成一个Callable对象。
public static Callable<Object> callable(Runnable task)
//当任务执行结束,Future.get()得到null
public static <T> Callable<T> callable(Runnable task, T result))
//当任务执行结束,Future.get()得到result对象。
ScheduledThreadPoolExecutor
详解
主要用来在给定的延迟之后运行任务,或者定期执行任务。ScheduledThreadPoolExecutor
的功能和Timer
类似,但ScheduledThreadPoolExecutor
功能更强大灵活,对应过个后台线程,Timer
对应的是单个后台线程。
运行机制
- 调用
ScheduledThreadPoolExecutor
的scheduleAtFixedRate()
或者scheduleWithFixedDelay()
方法是,会向ScheduledThreadPoolExecutor
的DelayQueue
添加一个实现了RunnableScheduledFutur
接口的ScheduledFutureTask
。DelayQueue
是无界的,所以maximunPoolSize
无效。DelayQueue
只能获取到时间已经到期的元素。 - 线程池中的线程从
DelayQueue
种获取ScheduledFutureTask
,然后执行任务。
ScheduleFutureTask
主要包含3个成员变量:
- long型成员变量time,表示这个任务将要执行的具体时间。
- long型成员变量
sequenceNumber
,表示这个任务在线程池中的序号。 - long型成员变量
period
,表示任务执行的间隔周期。
FutureTask
详解
Future接口和实现Future接口的FutureTask类,代表异步计算的结果。
FutureTask
的使用
可以把FutureTask
交给Executor
执行;也可以通过ExecutorService.submit(...)
方法返回一个FutureTask
,然后执行FutureTask.get()
方法或者FutureTask.cancel(...)
方法。除此以外,还可以单独使用FutureTask
。
package com.future;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
public class TutureTaskTest {
private final ConcurrentMap<Object, Future<String>> taskCache = new ConcurrentHashMap<Object, Future<String>>();
private String executionTask(final String taskName) throws ExecutionException, InterruptedException{
while(true){
Future<String> future = taskCache.get(taskName);
if(future == null){//没有该任务
Callable<String> task = new Callable<String>(){
@Override
public String call() throws Exception {
return taskName;
}
};
//创建任务
FutureTask<String> futureTask = new FutureTask<String>(task);
future = taskCache.putIfAbsent(taskName, futureTask);//如果这个key不存在就put返回null,否则不put并返回以前的值。
if(future == null){//put成功返回null
future = futureTask;
futureTask.run();//此次新添加了这个任务,执行这个任务
}
}
try{
return future.get();//等待该线程执行完任务之后返回
}catch(CancellationException e){
taskCache.remove(taskName, future);
}
}
}
public static void main(String[] args) {
}
}