Java&android ThreadPoolExcutor&ScheduledThreadPoolExecutor analyses of thread pool – Executor framework (multithreaded programming # 3)

The thread pool is essentially the same in Both Java and Android, so in this article we’ll take a look at the Thread pool Executor framework. Here are the main points:

1.Executor Framework Analysis

First of all, why do we need a thread pool? In Java, the creation and destruction of threads for asynchronous tasks requires some overhead. If we create a new thread for each task to execute, the creation and destruction of these threads will consume a lot of computing resources. Creating a new thread for each task to execute at the same time can cause an application under heavy load to crash. So the advent of thread pools is a promising solution to this problem. We will create several threads in the thread pool, when there is a task to execute when a thread from the thread pool to perform the task, if too much time task, beyond the number of threads of the thread pool, then the thread behind the task into a waiting queue to wait for, until there are threads in a thread pool free time to perform a task was obtained from the waiting queue for processing, This loop….. This greatly reduces the overhead of thread creation and destruction, and also relieves our application from being overloaded.

1.1 Two-level scheduling model of the Executor framework

A local operating system thread is created when a Java thread is started, and when the Java thread terminates, the operating system thread is also reclaimed. Each Java thread is mapped one-to-one to the threads of the local operating system, which schedules all threads and allocates them to available cpus. At the top, Java multithreaded programs divide applications into tasks and then map those tasks to a fixed number of threads using a user-level Executor framework. Underneath, the operating system kernel maps these threads to the hardware processor. Such two-level scheduling models are shown in the figure below:

As you can see from the figure, the application controls the upper level scheduling through the Executor framework, while the lower level scheduling is controlled by the operating system kernel, and the lower level scheduling is not controlled by the application.

1.2 Executor Framework Structure

The Executor framework structure consists of three main parts

1. Task: Includes the Runnable interface or Callable interface required by the task to be executed

2. Task execution: Includes Executor, the core interface of task execution mechanism, and EexcutorService interface inherited from Executor. Exrcutor two key class implements the ExecutorService interface (ThreadPoolExecutor and ScheduledThreadPoolExecutor).

3. Results of asynchronous computation: including the interface Future and the FutureTask class that implements the Future interface (explained in the next article)

Let’s take a look at the relationships between these classes using a UML diagram:

Extecutor is an interface that is the foundation of the Executor framework and separates the submission of tasks from their execution.

ThreadPoolExecutor is the core implementation class of the thread pool used to execute the submitted task.

ScheduledThreadPoolExecutor is an implementation class, can run in a given delay, or execute the command on a regular basis. ScheduledThreadPoolExecutor than the Timer is more flexible and more powerful.

The Future interface and the FutureTask class that implements the Future interface represent the result of asynchronous computation.

Runnable interface and Callable interface implementation class, or can be ThreadPoolExecutor ScheduledThreadPoolExecutor execution. The difference is that Runnable cannot return execution results, whereas Callable can return execution results.

Here is a diagram to understand the execution relationship between them:

Analysis:

The main thread first creates a Task object implementing the Runnable or Callable interface. The tool Executors can encapsulate a Runnable object into a Callable object in either of the following ways:

Executors. Callable (Runnable Task) or Object Resule (Runnable task).

You can then submit the Runnable object directly to the ExecutorService for execution, using the ExecutorService. Execute (Runnable Command) method. Or you can submit a Runnable or Callable object to the ExecutorService, The method can be executorService. submit(Runnable task) or executorService. submit(Callable task). Executorservice.submit (…) ExecutorService will return an object (FutureTask) that implements the Future interface. Of course, since FutureTask implements the Runnable interface, you can also create FutureTask directly and submit it to the ExecutorService for execution. Now that we have covered the main Executor framework architecture, and we have an overview of it, let’s focus on the two main thread pool implementation classes.

2. ThreadPoolExecutor analyses

ThreadPoolExecutor is a true implementation of threads created by using the factory Executors class. But the ThreadPoolExecutor constructor provides a set of parameters to configure the thread pool.

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }Copy the code

CorePoolSize: The number of core threads in the thread pool. By default, the number of core threads will remain alive in the thread pool even if they handle idle state. If the allowCoreThreadTimeOut property of ThreadPoolExecutor is set to true, idle core threads will execute a timeout policy while waiting for a new task to arrive, specified by keepAliveTime. When the wait time exceeds the keepAliveTime specified, the core thread is terminated.

MaximumPoolSize: The maximum number of threads that can be held in the thread pool. When the number of active threads reaches this value, new tasks will be blocked.

KeepAliveTime: Specifies the timeout period during which a non-core thread is idle before it is reclaimed. KeepAliveTime also applies to the core thread when the allowCoreThreadTimeOut property of ThreadPoolExecutor is set to true.

Unit: specifies the TimeUnit for keepAliveTime. This is an enumeration, usually using timeunit.milliseconds, timeunit.seconds, and timeunit.minutes.

WorkQueue: A queue of tasks in a thread pool in which Runnable objects submitted via the execute method of the thread pool are stored.

ThreadFactory: a threadFactory that provides the ability to create new threads for a thread pool. ThreadFactory is an interface that has only one method: Thread newThread (Runnable R).

RejectExecutionHandler, a less commonly used parameter, should be used when ThreadPoolExecutor is closed or when ThreadPoolExecutor is saturated (the maximum thread pool size is reached and the work queue is full). The execute method will notify the caller by calling the Handler’s rejectExecution method, which by default throws a RejectExecutionException. Now that we know the parameters of the relevant constructors, let’s look at the general rules for ThreadPoolExecutor to perform its tasks:

(1) If the number of thread pools has not reached the number of core threads, a core thread will be directly started to execute the task

(2) If the number of threads in the thread pool has reached or exceeded the number of core threads, the task will be inserted into the task queue and queued for execution.

(3) If the task cannot be inserted into the task queue in step (2), this is usually because the task queue is full. If the number of threads does not reach the maximum specified by the thread pool, a non-core thread will be immediately started to execute the task.

(4) If the number of threads has reached the maximum specified by the thread pool in step (3), the task will be rejected and ThreadPoolExecutor will call the rejectExecution method of the RejectExecutionHandler to notify the caller.

Now that we know the details of the configuration of ThreadPoolExecutor and the execution rules of ThreadPoolExecutor, let’s look at the three common thread pools. Each of these threads implements functionality directly or indirectly by configuring ThreadPoolExecutor. These three thread pools are FixedThreadPool, CachedThreadPool, ScheduledThreadPool and SingleThreadExecutor.

2.1 FixedThreadPool

The FixedThreadPool pattern uses a fixed number of priority threads to handle a number of tasks. A specified number of threads process all tasks, and once a thread has finished processing a task, it is used to process a new task (if any). The maximum number of threads in FixedThreadPool mode is fixed. The code to create a FixedThreadPool object is as follows:

ExecutorService fixedThreadPool=Executors.newFixedThreadPool(5);Copy the code

Let’s see
Create a FixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue());
    }Copy the code

FixedThreadPool corePoolSize and
The maximumPoolSize parameter is set to nThreads. When the number of threads in the thread pool is greater than
When corePoolSize, keepAliveTime is the maximum time for a non-core idle thread to wait for a new task, after which the non-core thread will be terminated, here
KeepAliveTime is set to 0L, indicating that non-core threads are terminated immediately. In fact, there are no non-core threads created, because the number of core threads is the same as the maximum number of threads.
Let’s look at how the execute() method of FixedThreadPool works:

Analysis:

(1) If the current number of running threads corePoolSize is small, create a new thread to execute the task.

(2) If the number of running threads in the current thread pool is equal to corePoolSize, subsequent tasks will be added to the LinkedBlockingQueue.

(3) After executing 1 in the figure, the thread will repeatedly fetch tasks from LinkedBlockingQueue in a loop to execute.

Another point to note here is that FixedThreadPool uses the unbounded queue LinkedBlockingQueue as the work queue of the thread pool (queue capacity integer.max_value). Using this queue as a work queue has the following effects on the thread pool

(1) After the number of threads in the current thread pool reaches corePoolSize, new tasks will wait in the unbounded queue.

(2) Since we are using an unbounded queue, the maximumPoolSize and keepAliveTime parameters are invalid.

(3) Due to the use of unbounded queues, a running FixedThreadPool does not reject tasks (unless shutdown and shutdownNow methods are executed). Therefore, the rejectExecution method of the RejectExecutionHandler is not called to throw an exception.

Here’s an example from a book on Java Programming Ideas:

public class LiftOff implements Runnable{ protected int countDown = 10; //Default private static int taskCount = 0; private final int id = taskCount++; public LiftOff() {} public LiftOff(int countDown) { this.countDown = countDown; } public String status() { return "#" + id + "(" + (countDown > 0 ? countDown : "LiftOff!" ) + ") "; } @Override public void run() { while(countDown-- > 0) { System.out.print(status()); Thread.yield(); }}}Copy the code

Declare a Runnable object using
FixedThreadPool performs the following tasks:

Public class FixedThreadPool {public static void main(String[] args) {// Three threads to perform five tasks ExecutorService exec = Executors.newFixedThreadPool(3); for(int i = 0; i < 5; i++) { exec.execute(new LiftOff()); } exec.shutdown(); }}Copy the code

CachedThreadPool first creates as many threads as necessary to execute a Task. As the program executes, some threads finish executing tasks and can be recycled before new threads are created to execute tasks. Creation method:

ExecutorService cachedThreadPool=Executors.newCachedThreadPool(); Copy the code

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue());
    }Copy the code

From this static method, we can see that
CorePoolSize of CachedThreadPool is set to 0, while
MaximumPoolSize is set
Integer. MAX_VALUE, namely
MaximumPoolSize is unbounded, and keepAliveTime is set to 60L in units of Max. That is, the maximum idle thread waiting time is 60 seconds, after which the thread will be terminated. And here it is
CachedThreadPool uses the SynchronousQueue, which has no capacity, as the work queue of the thread pool
MaximumPoolSize is unbounded, meaning that if the main thread submits tasks faster than
The speed at which a thread processes a task in maximumPoolSize
CachedThreadPool will constantly create new threads, and in extreme cases,
CachedThreadPool deplets CPU and memory resources by creating multiple threads.
CachedThreadPool
Execute () ¶

Analysis:

Synchronousqueue. offer(Runnable task) to add a task. If you have any idle thread is executing in current CachedThreadPool SynchronousQueue will. The poll (keepAliveTime, TimeUnit. NANOSECONDS), which is NANOSECONDS nanosecond is one over one billion of a second (is ms / 10 00), the main thread performs the offer operation and the poll operation performed by the idle thread are paired successfully, the main thread gives the task to the idle thread to execute, execute() method completes, otherwise step (2) is executed.

(2) when the initial number of threads CachedThreadPool is empty, or the current without the idle thread, there will be no thread to execute SynchronousQueue will. The poll (keepAliveTime, TimeUnit. NANOSECONDS). In this case, step (1) will fail, and CachedThreadPool will create a new thread to execute the task, and execute() will complete.

(3) in step (2) to create new threads will be after completion of task execution, will perform SynchronousQueue will. The poll (keepAliveTime, TimeUnit. NANOSECONDS), This poll operation causes the idle thread to wait in the SynchronousQueue for a maximum of 60 seconds. If the main thread submits a new task within 60 seconds, the idle thread will execute the new task submitted by the main thread; otherwise, the idle thread will be terminated. Since idle threads that are idle for 60 seconds are terminated, a CachedThreadPool that remains idle for a long time will not use any resources.

The SynchronousQueue is a blocking queue that has no capacity because the idle thread is removed at that time. Each insert operation must wait until a thread corresponds to it. The CachedThreadPool uses SynchronousQueue to delegate tasks from the main thread to idle threads. The process is as follows:

The use of CachedThreadPool
The case code is as follows:

public class CachedThreadPool { public static void main(String[] args) { ExecutorService exec = Executors.newCachedThreadPool(); for(int i = 0; i < 10; i++) { exec.execute(new LiftOff()); } exec.shutdown(); }}Copy the code

Creation method:

ExecutorService singleThreadExecutor=Executors.newSingleThreadExecutor(); Copy the code

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue()));
    }Copy the code

From the static method you can see that corePoolSize and maximumPoolSize for SingleThreadExecutor are set to 1, and the other parameters are the same as FixedThreadPool. The work queue used by SingleThreadExecutor is also the unbounded queue LinkedBlockingQueue. Due to the
SingleThreadExecutor’s adoption of unbounded queues has the same effect on the thread pool as FixedThreadPool, which is not described here. Similarly, let’s take a look at its operation process first:

Analysis:

(1) If the current number of threads is less than corePoolSize, that is, there are no threads running in the thread pool, then a new thread is created to execute the task.

(2) Add tasks to LinkedBlockingQueue when the number of threads in the thread pool is equal to corePoolSize.

(3) After the thread completes the task in (1), it repeatedly fetkes the task from the LinkedBlockingQueue in an infinite loop to execute.

The case code used by SingleThreadExecutor is as follows:

public class SingleThreadExecutor { public static void main(String[] args) { ExecutorService exec = Executors.newSingleThreadExecutor(); for (int i = 0; i < 2; i++) { exec.execute(new LiftOff()); }}}Copy the code

FixedThreadPool: Suitable for scenarios where the number of current threads needs to be limited to meet resource management requirements. It is suitable for heavily loaded servers.

SingleThreadExecutor: applies to tasks that need to be executed sequentially; And no more than one thread is active at any one point in time.

CachedThreadPool: An unbounded thread pool suitable for small programs that perform many short asynchronous tasks, or for lightly loaded servers.

3. ScheduledThreadPoolExecutor analyses

3.1 ScheduledThreadPoolExecutor execution mechanism analysis

ScheduledThreadPoolExecutor inherited from ThreadPoolExecutor. It is mainly used to execute tasks after a given delay or on a regular basis. ScheduledThreadPoolExecutor function is similar to the Timer, but are more powerful than the Timer, more flexible, the Timer is corresponding individual background threads, and can be specified in the constructor ScheduledThreadPoolExecutor more corresponding background threads. Next let’s look at ScheduledThreadPoolExecutor operating mechanism:

Analysis: DelayQueue is an unbounded queue, so the ThreadPoolExecutor maximumPoolSize in ScheduledThreadPoolExecutor meaningless. The implementation of ScheduledThreadPoolExecutor mainly divided into the following two parts

(1) when calling ScheduledThreadPoolExecutor scheduleAtFixedRate () method or scheduleWithFixedDelay () method, Will be added to ScheduledThreadPoolExecutor DelayQueue an implementation of a RunnableScheduledFuture ScheduleFutureTask interface.

(2) The thread in the thread pool retrieves the ScheduleFutureTask from DelayQueue and executes the task.

3.2 how to create ScheduledThreadPoolExecutor?

ScheduledThreadPoolExecutor usually use the factory class Executors to create, Executors can create two types of ScheduledThreadPoolExecutor, as follows:

ScheduledThreadPoolExecutor: (1) can perform parallel tasks, that is, multiple threads execute simultaneously.

(2) SingleThreadScheduledExecutor: can perform a single thread.

Create ScheduledThreadPoolExecutor method construction is as follows:

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory)Copy the code

create
SingleThreadScheduledExecutor
The method is constructed as follows:

public static ScheduledExecutorService newSingleThreadScheduledExecutor()
public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory)Copy the code

Create instance object code as follows:

ScheduledExecutorService scheduledThreadPoolExecutor=Executors.newScheduledThreadPool(5);Copy the code

ScheduledExecutorService singleThreadScheduledExecutor=Executors.newSingleThreadScheduledExecutor();Copy the code

ScheduledThreadPoolExecutor: suitable for multiple background threads execute periodic tasks, at the same time in order to meet the demand for resources management need to limit the number of background thread application scenario.

SingleThreadScheduledExecutor: suitable for need a single background threads execute cycle task, need to make sure that tasks at the same time sequential application scenario.

3.4 ScheduledThreadPoolExecutor use cases

We create a Runnable object, and then use the ScheduledThreadPoolExecutor Scheduled () to delay tasks, output execution time can be:

Let’s start with the method of delaying execution of this class:

public ScheduledFuture schedule(Runnable command,long delay, TimeUnit unit);Copy the code

Delay: How long to delay execution.

Unit: specifies the TimeUnit for keepAliveTime. This is an enumeration, usually using timeunit.milliseconds, timeunit.seconds, and timeunit.minutes.

Note here that this method returns an instance of ScheduledFuture, which can be used to get thread status information and delay time.

package com.zejian.Executor; import java.text.ParsePosition; import java.text.SimpleDateFormat; import java.util.Date; /** * @author zejian * @time * @decrition creates a WorkerThread to inherit Runnable */ public class WorkerThread implements Runnable{ @Override public void run() { System.out.println(Thread.currentThread().getName()+" Start. Time = "+getNowDate()); threadSleep(); System.out.println(Thread.currentThread().getName()+" End. Time = "+getNowDate()); } public void threadSleep(){try {thread.sleep (3000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); }} @return Return time type YYYY-MM-DD HH: MM :ss */ public static String getNowDate() {Date currentTime = new Date(); SimpleDateFormat formatter; formatter = new SimpleDateFormat ("yyyy-MM-dd HH:mm:ss"); String ctime = formatter.format(currentTime); return ctime; }}Copy the code

The execution classes are as follows:

package com.zejian.Executor; import java.text.ParsePosition; import java.text.SimpleDateFormat; import java.util.Date; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; /** * @author zejian * @time ** @decrition execution class */ public class ScheduledThreadPoolTest {public static void main(String[] args) { ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5); try { //schedule to run after sometime System.out.println("Current Time = "+getNowDate()); for(int i=0; i<3; i++){ Thread.sleep(1000); WorkerThread worker = new WorkerThread(); / / delay 5 SECONDS after the execution scheduledThreadPool. The schedule (worker, 10, TimeUnit. SECONDS); } Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } scheduledThreadPool.shutdown(); while(! scheduledThreadPool.isTerminated()){ //wait for all tasks to finish } System.out.println("Finished all threads"); } @return Return time type YYYY-MM-DD HH: MM :ss */ public static String getNowDate() {Date currentTime = new Date(); SimpleDateFormat formatter; formatter = new SimpleDateFormat ("yyyy-MM-dd HH:mm:ss"); String ctime = formatter.format(currentTime); return ctime; }}Copy the code

Run input Execution result:

Threaded tasks do start executing after a 10-second delay. This is where the schedule() method is used. Let’s talk about it againTwo available for periodic execution of tasksMethods.

public ScheduledFuture scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit)

The scheduleAtFixedRate method is used to periodically execute a given task after the initialDelay is over. The duration is period, where initialDelay is the initialDelay.

 public ScheduledFuture scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit);Copy the code

The implementation case code reference is given below:

package com.zejian.Executor; import java.util.Date; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** * @author zejian * @time * @decrition cycle function test class */ public class ScheduledTask {public ScheduledThreadPoolExecutor se = new ScheduledThreadPoolExecutor(5); public static void main(String[] args) { new ScheduledTask(); } public void fixedPeriodSchedule() {// Set the runnable periodschedule that can be periodable, with an initial delay of 0 and a task interval of 1 second for(int I =0; i<5; i++){ se.scheduleAtFixedRate(new FixedSchedule(), 0, 1, TimeUnit.SECONDS); } } public ScheduledTask() { fixedPeriodSchedule(); } class FixedSchedule implements Runnable {public void run() {system.out.println (" " "+ thread.currentThread ().getName()+" new Date(system.currentTimemillis ())); }}}Copy the code

Main reference books:

Java Core Technology Volume 1

Android development art exploration

The art of Concurrent programming in Java