Part of this article is excerpted from The Art of Concurrent Programming in Java

An overview of the

Thread pools in Java are the most common concurrency framework for running scenarios, and their proper use provides three benefits:

  • Reduce resource consumption. Reduce the cost of thread creation and destruction by reusing existing threads
  • Improve response speed. When a task arrives, it can execute immediately without waiting for a thread to be created
  • Improved thread manageability. Threads are scarce resources. Unified allocation, tuning, and monitoring by thread pools can reduce resource consumption and improve system stability

Implementation principle of thread pool

As you can see from the figure, when a new task is submitted to the thread pool, the thread pool process is as follows:

  1. The thread pool determines whether all the threads in the core thread pool are executing a task. If not, create a new worker thread to execute the task, or proceed to the next process
  2. The thread pool determines whether the work queue is full, if not, it stores the newly submitted task in the work queue, otherwise it goes to the next process
  3. The thread pool determines whether all the threads in the thread pool are working. If not, a new worker thread is created to perform the task, otherwise the saturation policy will handle the task

Using thread pools

1. Create a thread pool

We can create a thread pool using ThreadPoolExecutor

new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
Copy the code

To create a thread, enter several parameters as follows:

  • CorePoolSize (base size of thread pool)

    When a task is submitted to the thread pool, a thread is created to execute the task, even if other free base threads are able to execute new tasks, until the number of tasks that need to be executed exceeds the base size of the thread pool

  • MaximumPoolSize (maximum number of thread pools)

    The maximum number of threads allowed to be created by the thread pool. If the queue is full and the number of created threads is less than the maximum number, the thread pool creates a new thread to execute the task. It is worth noting that this parameter has no effect if the task queue is unbounded blocking

  • KeepAliveTime (thread activity hold time)

    The amount of time that a worker thread in a thread pool remains alive after it is idle. If there are many tasks and the execution time of each task is short, you can increase the execution time to improve thread utilization

  • Unit (How long a thread remains active)

    The options are DAYS, HOURS, MINUTES, MILLISECONDS, MICROSECONDS and NANOSECONDS

  • workQueue

    Blocking queues used to hold tasks waiting to be executed. You can choose from the following blocking queues:

    • ArrayBlockingQueue

      Is a bounded blocking queue based on an array structure that sorts elements according to FIFO (first-in, first-out)

    • LinkedBlockingQueue

      A linked list-based blocking queue that sorts elements by FIFO and typically has a higher throughput than an ArrayBlockingQueue

    • SynchronousQueue

      A blocking queue that does not store elements, each insert operation must wait until another thread calls the remove operation, otherwise the insert operation is consistently blocked, and throughput is usually higher than LinkedBlockingQueue

    • PriorityBlockingQueue

      An unbounded blocking queue with priority

  • threadFactory

    Used to set up a factory for creating threads, which can be used to give each created thread a more meaningful name

  • Handler (Saturation Strategy)

    When both the task and the thread pool are full, the thread pool is saturated and a policy must be adopted to handle submitted new tasks. The thread pool framework in JDK5 provides the following four strategies:

    • AbortPolicy: Directly throws an exception. This policy is adopted by default
    • CallerRunsPolicy: Runs the task using the caller’s thread
    • DiscardOldestPolicy: Discards the last task in the queue and executes the current task
    • DiscardPolicy: Do not process, discard

    You can also implement the RejectedExecutionHandler interface to customize a policy as required

2. Submit tasks to the thread pool

You can submit tasks to the thread pool using the execute() and submit() methods

  • The execute() method is used to submit tasks that do not require a return value, so there is no way to determine whether the task was successfully executed by the thread pool

    threadsPool.execute(new Runnable() {
    @Override
    public void run(a) {
        / /...}})Copy the code

  • The submit() method is used to submit tasks that require a return value, and the thread pool returns a Future object that can be used to determine whether the task was successfully executed

    Future<Object> future = executor.submit(hasReturnValueTask);
try {
    Object s = future.get();
} catch(InterruptedException e) {
    // Handle the interrupt exception
} catch(ExecutionException e) {
    // Handle the exception that tasks cannot be executed
} finally {
    // Close the thread pool
    executor.shutdown();
}
Copy the code

3. Close the thread pool

A thread pool can be shutdown by calling the shutdown or shutdownNow methods of the thread pool, which work by iterating through the worker threads in the pool and interrupting them one by one by calling the interrupt method, so tasks that cannot respond to interrupts may never be terminated

There are some differences between the shutdown method and shutdownNow method:

  • The shutdownNow method first sets the thread pool state to STOP, then attempts to STOP all threads executing or suspending tasks and returns a list of tasks awaiting execution
  • The shutdown method simply sets the thread pool state to shutdown and interrupts all threads that are not executing a task

The isShutdown method returns true whenever either of the two shutdown methods is called, and the thread pool is closed successfully when all tasks are closed, in which case the isTerminaed method returns true. Depending on the nature of the task submitted to the thread pool, the shutdown method is usually called to shutdown the thread pool, or the shutdownNow method can be called if the task is not expected to complete

Simple Web server based on thread pool technology

Current browsers all support multi-threaded access. For example, when a page is requested, static resources such as images contained in the page will be concurrently obtained by the browser. If a Web server is single-threaded, processing incoming requests sequentially will undoubtedly affect the user experience, so most Web servers support concurrent access

Let’s use a thread pool to construct a simple Web server that handles HTTP requests and currently can only handle simple text and image content. The Web server uses the main thread to continuously accept Socket connections from clients and submit connections and requests to the thread pool for processing, which enables the Web server to process requests from multiple clients simultaneously

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; }}/** * Start SimpleHttpServer */
    public static void start(a) throws Exception {
        serverSocket = new ServerSocket(port);
        Socket socket = null;
        while((socket = serverSocket.accept()) ! =null) {
            // Receive a client Socket, generate an HttpRequestHandler, and put it into the thread pool
            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(a) {
            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();
                // Calculate the absolute path
                String filePath = SimpleHttpServer.class.getResource(header.split("") [1]).getPath();
                out = new PrintWriter(socket.getOutputStream());
                // If the requested resource has a JPG or ico suffix, read the resource and print it
                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); }}}/** * close the stream or 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(); }}Copy the code