This article introduces the Adapter module of the Dynamic Thread Pool framework (DynamicTp). This module is mainly used for thread pool management of some third-party components, so that the built-in thread pool of third-party components can also enjoy dynamic parameter adjustment, alarm monitoring and other enhancements.
DynamicTp project address
At present, there are more than 500 stars. Thank you for your star. Welcome pr and contribute to open source in your spare time
Gitee address: gitee.com/yanhom/dyna…
Github address: github.com/lyh200/dyna…
series
Us group dynamic thread pool practice ideas, open source
Dynamic thread pool framework (DynamicTp), monitoring and source code parsing
The Adapter has been connected to the component
Adapter module has been connected to the thread pool management of the three webServers (Tomcat, Jetty and Undertow) built-in in SpringBoot. The implementation level is also decouple with the core module, and the event mechanism of Spring is used for notification monitoring processing.
You can see that there are two listeners
-
When a configuration change is detected in the configuration center, a RefreshEvent event will be issued after the internal thread pool of our project is updated, and DtpWebRefreshListener will update the thread pool parameters of the corresponding WebServer after the event is detected.
-
The same is true for monitoring alarms. When a monitoring task is performed in DtpMonitor, a CollectEvent event is released. After the DtpWebCollectListener listens to this event, it collects thread pool indicator data of the corresponding WebServer.
To manage thread pools for third-party components, you must first have some familiarity with the components, understand the entire process of a request, and find the thread pool that handles the request. These thread pools may not be the ThreadPoolExecutor class of the JUC package, but may also be the thread pool implemented by the component itself. But the basic principles are the same.
Tomcat, Jetty, and Undertow are all the same. Instead of using the thread pool implementation provided by JUC, they implemented their own, or extended, implementations of JUC. If not, we need to consider using reflection to retrieve the thread pool.
Implementation of Tomcat internal thread pool
- The Tomcat internal thread pool does not directly use the JUC ThreadPoolExecutor. Instead, it inherits the JUC Executor class and overrides the execute() method, which varies from version to version.
1. Inherits JUC native ThreadPoolExecutor (version 9.0.50 and below) and overwrites some methods, mainly execute() and afterExecute()
JUC AbstractExecutorService (version 9.0.51 and above). The code basically copies JUC’s ThreadPoolExecutor and tweaks the execute() method accordingly
Tomcat’s ThreadPoolExecutor class has the same name as JUC’s ThreadPoolExecutor class, and its execute() method is as follows:
public void execute(Runnable command, long timeout, TimeUnit unit) {
submittedCount.incrementAndGet();
try {
super.execute(command);
} catch (RejectedExecutionException rx) {
if (super.getQueue() instanceof TaskQueue) {
final TaskQueue queue = (TaskQueue)super.getQueue();
try {
if(! queue.force(command, timeout, unit)) { submittedCount.decrementAndGet();throw new RejectedExecutionException(sm.getString("threadPoolExecutor.queueFull")); }}catch (InterruptedException x) {
submittedCount.decrementAndGet();
throw newRejectedExecutionException(x); }}else {
submittedCount.decrementAndGet();
throwrx; }}}Copy the codeCould see that he is the first call of the parent class the execute () method, and then catch exceptions RejectedExecutionException, to judge if the type of the task queue is TaskQueue again, then try to add tasks to the task queue, if add fails, prove that the queue is full, and then execute rejection policies, Here submittedCount is an atomic variable that records the number of tasks submitted to this thread pool but not completed (mainly used in the offer() method of the TaskQueue mentioned below). Why? Keep reading!
- Tomcat defines a TaskQueue that inherits the self-linkedBlockingQueue, which essentially overwrites the offer() method.
@Override
public boolean offer(Runnable o) {
//we can't do any checks
if (parent==null) return super.offer(o);
//we are maxed out on threads, simply queue the object
if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);
//we have idle threads, just add it to the queue
if (parent.getSubmittedCount()<=(parent.getPoolSize())) return super.offer(o);
//if we have less threads than maximum force creation of a new thread
if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
//if we reached here, we need to add it to the queue
return super.offer(o);
}
Copy the codeYou can see that he made several judgments before joining the queue, where parent is the thread pool object to which he belongs
1. If parent is null, call the parent offer method to join the queue
2. If the current number of threads is equal to the maximum number of threads, call the parent class offer() method to join the queue
3. If the number of unexecuted tasks is less than or equal to the number of current threads, consider whether there are idle threads
If the current number of threads is less than the maximum number of threads, return false and go back to the JUC thread pool to see if you need to add new threads to perform the task
5. All other cases go straight to the team
- Since the Tomcat thread pool is mainly used to do IO tasks, the main purpose of all this is to better support IO intensive scenarios with minimal cost changes. The JUC thread pool is mainly suitable for CPU intensive scenarios. Recall the execution flow of the JUC native thread pool ThreadPoolExecutor#execute() method
1. If the current number of threads is smaller than the core thread pool, create a new thread to process the submitted task
2. If the number of current threads is greater than the number of core threads and the queue is not full, the task is put into the task queue for execution
3. If the number of current thread pools is larger than the core thread pool, smaller than the maximum number of threads, and the task queue is full, a new thread is created to execute the submitted task
4. If the current number of threads is equal to the maximum number of threads and the queue is full, the task is rejected
As you can see, when the number of current threads is greater than the number of core threads, the JUC native thread pool first queues tasks for execution, rather than creating threads for execution.
If Tomcat receives more requests than the number of core threads, the request is queued up for processing by the core thread, which slows down the overall processing speed of the request. Instead of using the JUC native thread pool, Tomcat uses the Offer () method of TaskQueue to modify the JUC thread pool execution flow. The Tomcat thread pool execution flow is as follows:
1. If the current number of threads is smaller than the core thread pool, create a new thread to process the submitted task
2. If the number of current thread pools is larger than the core thread pool and smaller than the maximum number of threads, a new thread is created to execute the submitted task
3. If the current number of threads is equal to the maximum number of threads, the task is put into a task queue for execution
4. If the queue is full, the denial policy is executed
- The Tomcat core thread pool can be obtained using the following methods
public Executor doGetTp(WebServer webServer) {
TomcatWebServer tomcatWebServer = (TomcatWebServer) webServer;
return tomcatWebServer.getTomcat().getConnector().getProtocolHandler().getExecutor();
}
Copy the code- To dynamically adjust the thread parameters of the Tomcat thread pool, you can add the following parameters to the configuration file after introducing the DynamicTp dependency. The parameters are the same as those of the Properties configuration class provided by SpringBoot. See readme for a complete example of the configuration file
spring:
dynamic:
tp:
// Other Configuration Items
tomcatTp: Tomcat Web Server thread pool configuration
minSpare: 100 Number of core threads
max: 400 # Maximum number of threads
Copy the codeTomcat thread pool is introduced here, through some of the above introduction, we must be very clear about the process of Tomcat thread pool task.
Implementation of Jetty’s internal thread pool
- The Jetty internal ThreadPool, which defines a ThreadPool top-level interface that inherits from Executor, has the following implementation classes
- Internal main use of QueuedThreadPool this implementation class, the thread pool execution process is not detailed explanation, interested can read the source code, the core idea is similar, around the number of core threads, the maximum number of threads, task queue three parameters start, compared with Tocmat, actually quite simple.
public void execute(Runnable job)
{
// Determine if we need to start a thread, use and idle thread or just queue this job
int startThread;
while (true)
{
// Get the atomic counts
long counts = _counts.get();
// Get the number of threads started (might not yet be running)
int threads = AtomicBiInteger.getHi(counts);
if (threads == Integer.MIN_VALUE)
throw new RejectedExecutionException(job.toString());
// Get the number of truly idle threads. This count is reduced by the
// job queue size so that any threads that are idle but are about to take
// a job from the queue are not counted.
int idle = AtomicBiInteger.getLo(counts);
// Start a thread if we have insufficient idle threads to meet demand
// and we are not at max threads.
startThread = (idle <= 0 && threads < _maxThreads) ? 1 : 0;
// The job will be run by an idle thread when available
if(! _counts.compareAndSet(counts, threads + startThread, idle + startThread -1))
continue;
break;
}
if(! _jobs.offer(job)) {// reverse our changes to _counts.
if (addCounts(-startThread, 1 - startThread))
LOG.warn("{} rejected {}".this, job);
throw new RejectedExecutionException(job.toString());
}
if (LOG.isDebugEnabled())
LOG.debug("queue {} startThread={}", job, startThread);
// Start a thread if one was needed
while (startThread-- > 0)
startThread();
}
Copy the code- The Jetty thread pool provides a public access method as follows
public Executor doGetTp(WebServer webServer) {
JettyWebServer jettyWebServer = (JettyWebServer) webServer;
return jettyWebServer.getServer().getThreadPool();
}
Copy the code- To dynamically adjust the thread parameters of the Jetty thread pool, you can add the following configuration to the configuration file after introducing the DynamicTp dependency. The parameters are named the same as the Properties configuration class provided by SpringBoot. See the readme project for a complete example of the configuration file
spring:
dynamic:
tp:
// Other Configuration Items
jettyTp: # Jetty Web Server thread pool configuration
min: 100 Number of core threads
max: 400 # Maximum number of threads
Copy the codeUndertow implementation of internal thread pools
-
Because of its strong performance and light weight, Undertow is still widely used now. Wildfly (former Jboss) has adopted Undertow as the default internal WebServer since 8, and Tomcat was used before. Those of you who know Undertow should know that he builds on the XNIO framework (pre-3.x), which is an excellent Java NIO-based networking framework developed by Jboss. When it comes to network programming, Netty is already the de facto standard, and the benefits of using Netty are far greater than what XNIO can offer. So let’s hope for 3.0, which was announced three years ago and hasn’t happened yet. The following introduction is based on Undertow 2.x
-
Undertow internally defines a top-level thread pool interface called TaskPool, which has several implementations as shown in the figure below. Each of these implementation classes is a combination of classes, internally maintaining a JUC Executor architecture class or an EnhancedQueueExecutor class provided by Jboss (which also inherits the JUC ExecutorService class), and the execution process can be analyzed by itself
- If not, create an internal class according to the configuration parameters, whether to use JUC’s ThreadPoolExecutor or Jboss’s EnhancedQueueExecutor
- The Undertow thread pool does not provide a public access method, so it does this by reflection
public Executor doGetTp(WebServer webServer) {
UndertowWebServer undertowWebServer = (UndertowWebServer) webServer;
Field undertowField = ReflectionUtils.findField(UndertowWebServer.class, "undertow");
if (Objects.isNull(undertowField)) {
return null;
}
ReflectionUtils.makeAccessible(undertowField);
Undertow undertow = (Undertow) ReflectionUtils.getField(undertowField, undertowWebServer);
if (Objects.isNull(undertow)) {
return null;
}
return undertow.getWorker();
}
Copy the code- To dynamically adjust the thread parameters of the Undertow thread pool, add the following configuration to the configuration file after introducing the DynamicTp dependency. See the readme project for a complete example of the configuration file
spring:
dynamic:
tp:
// Other Configuration Items
undertowTp: # Undertow Web Server thread pool configuration
coreWorkerThreads: 100 # worker Number of core threads
maxWorkerThreads: 400 # worker Maximum number of threads
workerKeepAlive: 60 # idle thread timeout
Copy the codeconclusion
The above introduces some situations of the built-in thread pool of Tomcat, Jetty and Undertow, with emphasis on Tomcat. The space is limited. You can analyze the other two if you are interested in them. It also introduces how to dynamically adjust thread pool parameters based on DynamicTp, which is very useful when we do WebServer performance tuning.
Again, we welcome you to use the DynamicTp framework and work together to improve the project.
ScheduledExecutorService can be used to monitor thread halt because of Tomcat version inconsistency during DynamicTp use.
To contact me
Welcome to add my wechat or follow the public account communication, together stronger!
Public id: CodeFox
WeChat: yanhom1314