Shulou(Shulou.com)06/02 Report--

Today, I will talk to you about what are the eight rejection strategies of Java thread pool ThreadPoolExecutor, which may not be well understood by many people. in order to make you understand better, the editor has summarized the following. I hope you can get something from this article.

Preface

When it comes to the thread pool of java, nothing is more familiar than the ExecutorService interface. The api under the new java.util.concurrent package of jdk1.5 greatly simplifies the development of multithreaded code. Whether you use FixedThreadPool or CachedThreadPool, the implementation behind it is ThreadPoolExecutor. ThreadPoolExecutor is a typical product of cache pooling design, because the pool has size, when the pool volume is not enough to carry, it involves a reject strategy. Four thread pool rejection policies have been preset in JDK. Let's talk more about the usage scenarios of these policies and which other rejection policies we can extend.

Pooling design idea

Chi dialect design should not be a new term. Our common implementations such as java thread pooling, jdbc connection pooling, and redis connection pooling are representative implementations of this kind of design. This design will initially preset resources, and the problem is to offset the consumption of resources each time, such as the overhead of creating threads, obtaining remote connections, and so on. Just like when you go to the cafeteria to eat, the aunt will put several meals there first, and when you come, you can just take the lunch box and add food. You no longer have to serve rice and vegetables temporarily, and it will be more efficient. In addition to initializing resources, pooling design also includes the following characteristics: the initial value of the pool, the activity value of the pool, the maximum value of the pool, and so on, which can be directly mapped to the member properties of the java thread pool and the database connection pool.

The time when the thread pool triggers the reject policy

Unlike the data source connection pool, the thread pool has an additional blocking queue to buffer in addition to the initial size and pool maximum. Data source connection pooling generally triggers a reject policy when the number of connections requested exceeds the maximum of the connection pool. The policy is to block the waiting time or throw an exception directly. The trigger time of the thread pool is as follows:

As shown in the figure, if you want to know when the thread pool triggers rejection roughly, you need to make clear the specific meaning of the above three parameters, which is the result of the overall coordination of these three parameters, rather than simply exceeding the maximum number of threads will trigger thread rejection rough. When the number of tasks submitted is greater than corePoolSize, it will be prioritized to the queue buffer. Only when the buffer is filled will you judge whether the currently running task is greater than maxPoolSize. When it is less than, a new thread processing is created. The reject policy is triggered when the number of tasks submitted is greater than (maxPoolSize + queueCapacity). In summary, the reject policy of the thread pool is triggered when the number of submitted tasks is greater than (reject + reject).

JDK built-in four kinds of thread pool reject policy interface definition

Before analyzing the thread pool rejection policy that comes with JDK, take a look at the deny policy API defined by JDK, as follows:

Public interface RejectedExecutionHandler {void rejectedExecution (Runnable r, ThreadPoolExecutor executor);}

The definition of the API is very clear. When the rejection policy is triggered, the thread pool will call the specific policy you set and pass the currently submitted task and the thread pool instance itself to you for processing. Different scenarios will have different considerations for how to handle it. Let's see which implementations are built into JDK for us:

CallerRunsPolicy (caller running policy) public static class CallerRunsPolicy implements RejectedExecutionHandler {public CallerRunsPolicy () {} public void rejectedExecution (Runnable r, ThreadPoolExecutor e) {if (! e.isShutdown ()) {r.run ();}

Function: when a deny policy is triggered, it is handled by the current thread that submits the task as long as the thread pool is not closed.

Usage scenarios: it is generally used in scenarios where failures are not allowed, performance requirements are low, and concurrency is small, because thread pools are generally not closed, that is, submitted tasks must be run. However, since the tasks are executed by the caller thread, when the tasks are submitted multiple times, the execution of subsequent tasks will be blocked, and the performance and efficiency will naturally slow down.

AbortPolicy (abort Policy) public static class AbortPolicy implements RejectedExecutionHandler {public AbortPolicy () {} public void rejectedExecution (Runnable r, ThreadPoolExecutor e) {throw new RejectedExecutionException ("Task" + r.toString () + "rejected from" + e.toString ());}}

Function: when a reject policy is triggered, an exception that rejects execution is thrown directly. The abort policy means to interrupt the current execution process.

Usage scenario: there is no special scenario for this, but the exception thrown should be handled correctly. The default policy in ThreadPoolExecutor is the series ThreadPoolExecutor of the AbortPolicy,ExecutorService interface, because there is no rejection policy displayed, so the default is this. Note, however, that the thread pool instance queues in ExecutorService are unbounded, meaning that bursting memory does not trigger a reject policy. When you customize a thread pool instance, you must use this policy to handle the exception that is thrown when the policy is triggered, because it will interrupt the current execution process.

DiscardPolicy (discard Policy) public static class DiscardPolicy implements RejectedExecutionHandler {public DiscardPolicy () {} public void rejectedExecution (Runnable r, ThreadPoolExecutor e) {}}

Function: drop this task directly and quietly without triggering any action

Usage scenario: if the task you submit is not important, you can use it. Because it is an empty realization, will quietly engulf your task. So this strategy is basically out of use.

DiscardOldestPolicy public static class DiscardOldestPolicy implements RejectedExecutionHandler {public DiscardOldestPolicy () {} public void rejectedExecution (Runnable r, ThreadPoolExecutor e) {if (! e.isShutdown ()) {e.getQueue () .poll (); e.execute (r);}

Function: if the thread pool is not closed, pop up the element in the queue header and then try to execute the

Usage scenario: this strategy still discards tasks and is silent when they are discarded, but it is characterized by discarding old unexecuted tasks and tasks with a higher priority to be executed. Based on this feature, the only scenario I can think of is to publish and modify the message. When the message is released, it has not been executed, and the updated message comes again. At this time, the version of the unexecuted message is lower than the currently submitted version of the message and can be discarded. Because there may be messages with a lower version in the queue that will be queued for execution, be sure to compare the version of the message when you actually process the message.

Thread rejection policy public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {protected static final Logger logger = LoggerFactory.getLogger (AbortPolicyWithReport.class) in dubbo implemented by third parties; private final String threadName; private final URL url; private static volatile long lastPrintTime = 0; private static Semaphore guard = new Semaphore (1); public AbortPolicyWithReport (String threadName, URL url) {this.threadName = threadName; this.url = url } @ Override public void rejectedExecution (Runnable r, ThreadPoolExecutor e) {String msg = String.format ("Thread pool is EXHAUSTED!" + "Thread Name:% s, Pool Size:% d (active:% d, core:% d, max:% d, largest:% d), Task:% d (completed:% d)," + "Executor status: (isShutdown:%s IsTerminated:%s, isTerminating:%s), in% s://%s:%d!, threadName, e.getPoolSize (), e.getActiveCount (), e.getCorePoolSize (), e.getMaximumPoolSize (), e.getLargestPoolSize (), e.getTaskCount (), e.getCompletedTaskCount (), e.isShutdown (), e.isTerminated (), e.isTerminating (), url.getProtocol (), url.getIp (), url.getPort ()) Logger.warn (msg); dumpJStack (); throw new RejectedExecutionException (msg);} private void dumpJStack () {/ / omitted implementation}}

As you can see, when the worker thread of dubbo triggers the thread rejection, it mainly does three things, and the principle is to let the user know the real reason for triggering the thread rejection policy as far as possible.

A warning-level log is output with detailed setting parameters for the thread pool, as well as the current state of the thread pool, as well as some details about the current deny task. It can be said that this log is more or less seen by those who have experience in production, operation and maintenance using dubbo. This log is simply a model of log printing, and other examples of log printing are spring. Thanks to such a detailed log, it is easy to locate the problem

Output the current thread stack details, this is very useful, when you can not locate the problem through the log information above, the dump thread context information at the crime scene is the lifesaver to find the problem. You can refer to the "Dubbo thread pool exhaustion event -"disaster caused by CyclicBarrier".

Continue to throw a deny execution exception, which makes this task fail, which inherits the feature of JDK's default reject policy.

Thread pool rejection policy in Netty private static final class NewThreadRunsPolicy implements RejectedExecutionHandler {NewThreadRunsPolicy () {super ();} public void rejectedExecution (Runnable r, ThreadPoolExecutor executor) {try {final Thread t = new Thread (r, "Temporary task executor"); t.start () } catch (Throwable e) {throw new RejectedExecutionException ("Failed to start a new thread", e);}

The implementation in Netty is much like CallerRunsPolicy in JDK and is reluctant to discard tasks. The difference is that CallerRunsPolicy is a task that is performed directly on the caller thread. Netty has created a new thread to process it. Therefore, the implementation of Netty can be extended to support efficient and high-performance scenarios compared to the use of policy enforcement by callers. However, it should also be noted that in the implementation of Netty, no judgment constraint is made when creating a thread, that is, as long as the system has resources, a new thread will be created to handle it, and the exception of thread creation failure will not be thrown until there are no new threads in the new.

Thread pool rejection policy in activeMq new RejectedExecutionHandler () {@ Override public void rejectedExecution (final Runnable r, final ThreadPoolExecutor executor) {try {executor.getQueue () .offer (r, 60, TimeUnit.SECONDS);} catch (InterruptedException e) {throw new RejectedExecutionException ("Interrupted waiting for BrokerService.worker") } throw new RejectedExecutionException ("Timed Out while attempting to enqueue Task.")

The policy in activeMq belongs to the best effort execution policy, and when the reject policy is triggered, it will try again for one minute. Re-shove the task into the task queue and throw an exception when the one-minute timeout is not successful

Thread pool rejection policy in pinpoint public class RejectedExecutionHandlerChain implements RejectedExecutionHandler {private final RejectedExecutionHandler [] handlerChain; public static RejectedExecutionHandler build (List chain) {Objects.requireNonNull (chain, "handlerChain must not be null"); RejectedExecutionHandler [] handlerChain = chain.toArray (new RejectedExecutionHandler [0]); return new RejectedExecutionHandlerChain (handlerChain);} private RejectedExecutionHandlerChain (RejectedExecutionHandler [] handlerChain) {this.handlerChain = Objects.requireNonNull (handlerChain, "handlerChain must not be null") } @ Override public void rejectedExecution (Runnable r, ThreadPoolExecutor executor) {for (RejectedExecutionHandler rejectedExecutionHandler: handlerChain) {rejectedExecutionHandler.rejectedExecution (r, executor);}

The rejection policy implementation of pinpoint is very characteristic, which is different from other implementations. He defines a reject policy chain and wraps a list of reject policies. When a reject policy is triggered, the rejectedExecution in the policy chain is executed in turn.

Conclusion

The definition of java thread pool rejection policy interface is introduced from the thread pool design idea and the time when the thread pool triggers the reject policy. Supplemented by the definition of rejection policy of four kinds of JDK and four third-party open source software, this paper describes various ideas and usage scenarios of thread pool rejection policy implementation.

After reading the above, do you have any further understanding of the eight rejection strategies of Java thread pool ThreadPoolExecutor? If you want to know more knowledge or related content, please follow the industry information channel, thank you for your support.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.