本文部分摘自《Java 并发编程的艺术》
概述
Java 中的线程池是运行场景最多的并发框架,合理使用线程池能够带来三个好处:
- 降低资源消耗。通过重复利用已有的线程降低线程创建和销毁造成的消耗
- 提高响应速度。当任务到达时,任务可以不需要等待线程创建就能立即执行
- 提高线程可管理性。线程是稀缺资源,使用线程池进行统一分配、调优和监控,可以降低资源消耗,提高系统稳定性
线程池的实现原理
从图中可以看到,当提交一个新任务到线程池时,线程池的处理流程如下:
- 线程池判断核心线程池里的线程是否都在执行任务,如果不是,创建一个新的工作线程执行任务,否则进入下一流程
- 线程池判断工作队列是否已满,如果工作队列没有满,将新提交的任务存储在工作队列中,否则进入下一流程
- 线程池判断线程池里的线程是否都处于工作状态,如果没有,创建一个新的工作线程执行任务,否则交给饱和策略来处理这个任务
使用线程池
1. 创建线程池
我们可以通过 ThreadPoolExecutor 来创建一个线程池
new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
创建一个线程需要输入几个参数,如下:
-
corePoolSize(线程池的基本大小)
当提交一个任务到线程池时,线程池会创建一个线程来执行任务,即时其他空闲的基本线程能够执行新任务也会创建线程,等到需要执行的任务数大于线程池基本大小时就不再创建
-
maximumPoolSize(线程池最大数量)
线程池允许创建的最大线程数,如果队列满了,并且已创建的线程数小于最大线程数,则线程池会再创建新的线程执行任务。值得注意的是,如果使用无界阻塞队列做任务队列,则这个参数没有什么效果
-
keepAliveTime(线程活动保持时间)
线程池的工作线程空闲后,保持存活的时间。如果任务很多,并且每个任务的执行时间都比较短,可以调大时间,提高线程利用率
-
unit(线程保持活动时间的单位)
可选的单位有天(DAYS)、小时(HOURS)、分钟(MINUTES)、毫秒(MILLISECONDS)、微妙(MICROSECONDS)和纳秒(NANOSECONDS)
-
workQueue(任务队列)
用于保存等到执行的任务的阻塞队列,可以选择以下几个阻塞队列:
-
ArrayBlockingQueue
是一个基于数组结构的有界阻塞队列,此队列按 FIFO(先进先出)原则对元素进行排序
-
LinkedBlockingQueue
一个基于链表结构的阻塞队列,此队列按 FIFO 排序元素,吞吐量通常高于 ArrayBlockingQueue
-
SynchronousQueue
一个不存储元素的阻塞队列,每个插入操作必须等到另一个线程调用移除操作,否则插入操作一致处于阻塞状态,吞吐量通常要高于 LinkedBlockingQueue
-
PriorityBlockingQueue
一个具有优先级的无界阻塞队列
-
-
threadFactory
用于设置创建线程的工厂,可以通过线程工厂给每个创建出来的线程设置更有意义的名字
-
handler(饱和策略)
当任务和线程池都满了,说明线程池处于饱和状态,必须采取一种策略处理提交的新任务。在 JDK5 中线程池框架提供了以下四种策略:
- AbortPolicy:直接抛出异常,默认采取这种策略
- CallerRunsPolicy:使用调用者所在线程来运行任务
- DiscardOldestPolicy:丢弃队列最近的一个任务,并执行当前任务
- DiscardPolicy:不处理,丢弃掉
也可以根据需要实现 RejectedExecutionHandler 接口自定义策略
2. 向线程池提交任务
可以使用 execute() 和 submit() 方法向线程池提交任务
-
execute() 方法用于提交不需要返回值的任务,所以无法判断任务是否被线程池执行成功
threadsPool.execute(new Runnable() { @Override public void run() { //... } })
-
submit() 方法用于提交需要返回值的任务,线程池会返回一个 future 对象,通过这个对象可以判断任务是否执行成功
Future<Object> future = executor.submit(hasReturnValueTask); try { Object s = future.get(); } catch(InterruptedException e) { // 处理中断异常 } catch(ExecutionException e) { // 处理无法执行任务异常 } finally { // 关闭线程池 executor.shutdown(); }
3. 关闭线程池
可以通过调用线程池的 shutdown 或 shutdownNow 方法来关闭线程池,它们的原理是遍历线程池中的工作线程,逐个调用线程的 interrupt 方法来中断线程,所以无法响应中断的任务可能永远无法终止
shutdown 方法和 shutdownNow 方法存在一定的区别:
- shutdownNow 方法首先将线程池状态设置成 STOP,然后尝试停止所有正在执行或暂停任务的线程,并返回等待执行任务的列表
- shutdown 方法只是将线程池状态设置成 SHUTDOWN 状态,然后中断所有没有正在执行任务的线程
只要调用了这两个关闭方法中的任意一个,isShutdown 方法就会返回 true,当所有任务都已关闭,才表示线程池关闭成功,这时调用 isTerminaed 方法会返回 true。至于应该采用哪种方法关闭线程池,应该由提交到线程池的任务特性决定,通常调用 shutdown 方法关闭线程池,如果任务不一定要执行完成,可以调用 shutdownNow 方法
基于线程池技术的简单 Web 服务器
目前的浏览器都支持多线程访问,比如请求一个页面的时候,页面包含的图片等静态资源会被浏览器并发的获取。如果 Web 服务器是单线程的,按顺序处理发送过来的请求,无疑会影响用户体验,因此大部分 Web 服务器都支持并发访问
下面使用线程池来构造一个简单的 Web 服务器,这个 Web 服务器用来处理 HTTP 请求,目前只能处理简单的文本和图片内容。该 Web 服务器使用 main 线程不断接受客户端的 Socket 连接,将连接以及请求提交给线程池处理,这样使得 Web 服务器能同时处理多个客户端的请求
public class SimpleHttpServer {
static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
5, 10, 60L,
TimeUnit.SECONDS, new LinkedBlockingQueue<>(),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy());
static ServerSocket serverSocket;
static int port = 8080;
public static void setPort(int port) {
if (port > 0) {
SimpleHttpServer.port = port;
}
}
/**
* 启动 SimpleHttpServer
*/
public static void start() throws Exception {
serverSocket = new ServerSocket(port);
Socket socket = null;
while ((socket = serverSocket.accept()) != null) {
// 接收一个客户端Socket,生成一个HttpRequestHandler,放入线程池执行
threadPool.execute(new HttpRequestHandler(socket));
}
serverSocket.close();
}
static class HttpRequestHandler implements Runnable {
private Socket socket;
public HttpRequestHandler(Socket socket) {
this.socket = socket;
}
@Override
public void run() {
String line;
BufferedReader br = null;
BufferedReader reader = null;
PrintWriter out = null;
InputStream in = null;
try {
reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
String header = reader.readLine();
// 计算绝对路径
String filePath = SimpleHttpServer.class.getResource(header.split(" ")[1]).getPath();
out = new PrintWriter(socket.getOutputStream());
// 如果请求资源的后缀为 jpg 或 ico,则读取资源并输出
if (filePath.endsWith("jpg") || filePath.endsWith("ico")) {
in = new FileInputStream(filePath);
ByteArrayOutputStream baos = new ByteArrayOutputStream();
int i;
while ((i = in.read()) != -1) {
baos.write(i);
}
byte[] array = baos.toByteArray();
out.println("HTTP/1.1 200 OK");
out.println("Server: YeeQ");
out.println("Content-Type: image/jpeg");
out.println("Content-Length: " + array.length);
out.println("");
socket.getOutputStream().write(array, 0, array.length);
} else {
br = new BufferedReader(new InputStreamReader(new FileInputStream(filePath)));
out = new PrintWriter(socket.getOutputStream());
out.println("HTTP/1.1 200 OK");
out.println("Server: YeeQ");
out.println("Content-Type: text/html; charset=UTF-8");
out.println("");
while ((line = br.readLine()) != null) {
out.println(line);
}
}
out.flush();
} catch (Exception e) {
if (out != null) {
out.println("HTTP/1.1 500");
out.println("");
out.flush();
}
} finally {
close(br, in, reader, out, socket);
}
}
}
/**
* 关闭流或者socket
*/
private static void close(Closeable... closeables) {
if (closeables != null) {
for (Closeable closeable : closeables) {
if (closeable != null) {
try {
closeable.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
}
public static void main(String[] args) throws Exception {
SimpleHttpServer.start();
}
}