Java 线程池详解

2年前 (2022) 程序员胖胖胖虎阿
236 0 0

本文部分摘自《Java 并发编程的艺术》

概述

Java 中的线程池是运行场景最多的并发框架,合理使用线程池能够带来三个好处:

  • 降低资源消耗。通过重复利用已有的线程降低线程创建和销毁造成的消耗
  • 提高响应速度。当任务到达时,任务可以不需要等待线程创建就能立即执行
  • 提高线程可管理性。线程是稀缺资源,使用线程池进行统一分配、调优和监控,可以降低资源消耗,提高系统稳定性

线程池的实现原理

Java 线程池详解

从图中可以看到,当提交一个新任务到线程池时,线程池的处理流程如下:

  1. 线程池判断核心线程池里的线程是否都在执行任务,如果不是,创建一个新的工作线程执行任务,否则进入下一流程
  2. 线程池判断工作队列是否已满,如果工作队列没有满,将新提交的任务存储在工作队列中,否则进入下一流程
  3. 线程池判断线程池里的线程是否都处于工作状态,如果没有,创建一个新的工作线程执行任务,否则交给饱和策略来处理这个任务

使用线程池

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();
    }
}

版权声明:程序员胖胖胖虎阿 发表于 2022年10月13日 上午11:48。
转载请注明:Java 线程池详解 | 胖虎的工具箱-编程导航

相关文章

暂无评论

暂无评论...