18154768944

Committed 01 Oct 2025 07:26AM UTC coverage: 81.882% (-0.005%) from 81.887%

Build # 18154768944

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #290 from TAKETODAY/dev/jspecify

jspecify

Coverage Stats

59788 of 78013 branches covered (76.64%)

Branch coverage included in aggregate %.

141239 of 167496 relevant lines covered (84.32%)

3.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.62

today-context/src/main/java/infra/scheduling/concurrent/ThreadPoolTaskExecutor.java

/*
 * Copyright 2017 - 2025 the original author or authors.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see [https://www.gnu.org/licenses/]
 */

package infra.scheduling.concurrent;

import org.jspecify.annotations.Nullable;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import infra.core.task.AsyncTaskExecutor;
import infra.core.task.TaskDecorator;
import infra.core.task.TaskExecutor;
import infra.core.task.TaskRejectedException;
import infra.lang.Assert;
import infra.scheduling.SchedulingTaskExecutor;
import infra.util.ConcurrentReferenceHashMap;
import infra.util.concurrent.Future;

/**
 * JavaBean that allows for configuring a {@link ThreadPoolExecutor}
 * in bean style (through its "corePoolSize", "maxPoolSize", "keepAliveSeconds", "queueCapacity"
 * properties) and exposing it as a Infra {@link TaskExecutor}.
 * This class is also well suited for management and monitoring (e.g. through JMX),
 * providing several useful attributes: "corePoolSize", "maxPoolSize", "keepAliveSeconds"
 * (all supporting updates at runtime); "poolSize", "activeCount" (for introspection only).
 *
 * <p>The default configuration is a core pool size of 1, with unlimited max pool size
 * and unlimited queue capacity. This is roughly equivalent to
 * {@link java.util.concurrent.Executors#newSingleThreadExecutor()}, sharing a single
 * thread for all tasks. Setting {@link #setQueueCapacity "queueCapacity"} to 0 mimics
 * {@link java.util.concurrent.Executors#newCachedThreadPool()}, with immediate scaling
 * of threads in the pool to a potentially very high number. Consider also setting a
 * {@link #setMaxPoolSize "maxPoolSize"} at that point, as well as possibly a higher
 * {@link #setCorePoolSize "corePoolSize"} (see also the
 * {@link #setAllowCoreThreadTimeOut "allowCoreThreadTimeOut"} mode of scaling).
 *
 * <p><b>NOTE:</b> This class implements Infra
 * {@link TaskExecutor} interface as well as the
 * {@link java.util.concurrent.Executor} interface, with the former being the primary
 * interface, the other just serving as secondary convenience. For this reason, the
 * exception handling follows the TaskExecutor contract rather than the Executor contract,
 * in particular regarding the {@link TaskRejectedException}.
 *
 * <p>For an alternative, you may set up a ThreadPoolExecutor instance directly using
 * constructor injection, or use a factory method definition that points to the
 * {@link java.util.concurrent.Executors} class. To expose such a raw Executor as a
 * Infra {@link TaskExecutor}, simply wrap it with a
 * {@link ConcurrentTaskExecutor} adapter.
 *
 * @author Juergen Hoeller
 * @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
 * @see TaskExecutor
 * @see ThreadPoolExecutor
 * @see ThreadPoolExecutorFactoryBean
 * @see ConcurrentTaskExecutor
 * @since 4.0
 */
@SuppressWarnings("serial")
public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport implements AsyncTaskExecutor, SchedulingTaskExecutor {

  private final Object poolSizeMonitor = new Object();

  private int corePoolSize = 1;

  private int maxPoolSize = Integer.MAX_VALUE;

  private int keepAliveSeconds = 60;

  private int queueCapacity = Integer.MAX_VALUE;

  private boolean allowCoreThreadTimeOut = false;

  private boolean prestartAllCoreThreads = false;

  private boolean strictEarlyShutdown = false;

  @Nullable
  private TaskDecorator taskDecorator;

  @Nullable
  private ThreadPoolExecutor threadPoolExecutor;

  // Runnable decorator to user-level FutureTask, if different
  private final ConcurrentReferenceHashMap<Runnable, Object> decoratedTaskMap =
          new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);

  /**
   * Set the ThreadPoolExecutor's core pool size.
   * Default is 1.
   * <p><b>This setting can be modified at runtime, for example through JMX.</b>
   */
  public void setCorePoolSize(int corePoolSize) {
    synchronized(this.poolSizeMonitor) {
      if (this.threadPoolExecutor != null) {
        this.threadPoolExecutor.setCorePoolSize(corePoolSize);
      }
      this.corePoolSize = corePoolSize;
    }
  }

  /**
   * Return the ThreadPoolExecutor's core pool size.
   */
  public int getCorePoolSize() {
    synchronized(this.poolSizeMonitor) {
      return this.corePoolSize;
    }
  }

  /**
   * Set the ThreadPoolExecutor's maximum pool size.
   * Default is {@code Integer.MAX_VALUE}.
   * <p><b>This setting can be modified at runtime, for example through JMX.</b>
   */
  public void setMaxPoolSize(int maxPoolSize) {
    synchronized(this.poolSizeMonitor) {
      if (this.threadPoolExecutor != null) {
        this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);
      }
      this.maxPoolSize = maxPoolSize;
    }
  }

  /**
   * Return the ThreadPoolExecutor's maximum pool size.
   */
  public int getMaxPoolSize() {
    synchronized(this.poolSizeMonitor) {
      return this.maxPoolSize;
    }
  }

  /**
   * Set the ThreadPoolExecutor's keep-alive seconds.
   * <p>Default is 60.
   * <p><b>This setting can be modified at runtime, for example through JMX.</b>
   */
  public void setKeepAliveSeconds(int keepAliveSeconds) {
    synchronized(this.poolSizeMonitor) {
      if (this.threadPoolExecutor != null) {
        this.threadPoolExecutor.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);
      }
      this.keepAliveSeconds = keepAliveSeconds;
    }
  }

  /**
   * Return the ThreadPoolExecutor's keep-alive seconds.
   */
  public int getKeepAliveSeconds() {
    synchronized(this.poolSizeMonitor) {
      return this.keepAliveSeconds;
    }
  }

  /**
   * Set the capacity for the ThreadPoolExecutor's BlockingQueue.
   * <p>Default is {@code Integer.MAX_VALUE}.
   * <p>Any positive value will lead to a LinkedBlockingQueue instance;
   * any other value will lead to a SynchronousQueue instance.
   *
   * @see LinkedBlockingQueue
   * @see SynchronousQueue
   */
  public void setQueueCapacity(int queueCapacity) {
    this.queueCapacity = queueCapacity;
  }

  /**
   * Return the capacity for the ThreadPoolExecutor's BlockingQueue.
   *
   * @see #setQueueCapacity(int)
   */
  public int getQueueCapacity() {
    return this.queueCapacity;
  }

  /**
   * Specify whether to allow core threads to time out. This enables dynamic
   * growing and shrinking even in combination with a non-zero queue (since
   * the max pool size will only grow once the queue is full).
   * <p>Default is "false".
   *
   * @see ThreadPoolExecutor#allowCoreThreadTimeOut(boolean)
   */
  public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
    this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;
  }

  /**
   * Specify whether to start all core threads, causing them to idly wait for work.
   * <p>Default is "false", starting threads and adding them to the pool on demand.
   *
   * @see ThreadPoolExecutor#prestartAllCoreThreads
   */
  public void setPrestartAllCoreThreads(boolean prestartAllCoreThreads) {
    this.prestartAllCoreThreads = prestartAllCoreThreads;
  }

  /**
   * Specify whether to initiate an early shutdown signal on context close,
   * disposing all idle threads and rejecting further task submissions.
   * <p>By default, existing tasks will be allowed to complete within the
   * coordinated lifecycle stop phase in any case. This setting just controls
   * whether an explicit {@link ThreadPoolExecutor#shutdown()} call will be
   * triggered on context close, rejecting task submissions after that point.
   * <p>the default is "false", leniently allowing for late tasks
   * to arrive after context close, still participating in the lifecycle stop
   * phase. Note that this differs from {@link #setAcceptTasksAfterContextClose}
   * which completely bypasses the coordinated lifecycle stop phase, with no
   * explicit waiting for the completion of existing tasks at all.
   * <p>Switch this to "true" for a strict early shutdown signal analogous to
   * the 4.0-established default behavior of {@link ThreadPoolTaskScheduler}.
   * Note that the related flags {@link #setAcceptTasksAfterContextClose} and
   * {@link #setWaitForTasksToCompleteOnShutdown} will override this setting,
   * leading to a late shutdown without a coordinated lifecycle stop phase.
   *
   * @see #initiateShutdown()
   */
  public void setStrictEarlyShutdown(boolean defaultEarlyShutdown) {
    this.strictEarlyShutdown = defaultEarlyShutdown;
  }

  /**
   * Specify a custom {@link TaskDecorator} to be applied to any {@link Runnable}
   * about to be executed.
   * <p>Note that such a decorator is not necessarily being applied to the
   * user-supplied {@code Runnable}/{@code Callable} but rather to the actual
   * execution callback (which may be a wrapper around the user-supplied task).
   * <p>The primary use case is to set some execution context around the task's
   * invocation, or to provide some monitoring/statistics for task execution.
   * <p><b>NOTE:</b> Exception handling in {@code TaskDecorator} implementations
   * is limited to plain {@code Runnable} execution via {@code execute} calls.
   * In case of {@code #submit} calls, the exposed {@code Runnable} will be a
   * {@code FutureTask} which does not propagate any exceptions; you might
   * have to cast it and call {@code Future#get} to evaluate exceptions.
   * See the {@code ThreadPoolExecutor#afterExecute} javadoc for an example
   * of how to access exceptions in such a {@code Future} case.
   */
  public void setTaskDecorator(TaskDecorator taskDecorator) {
    this.taskDecorator = taskDecorator;
  }

  /**
   * Note: This method exposes an {@link ExecutorService} to its base class
   * but stores the actual {@link ThreadPoolExecutor} handle internally.
   * Do not override this method for replacing the executor, rather just for
   * decorating its {@code ExecutorService} handle or storing custom state.
   */
  @Override
  protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedHandler) {
    BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);

    ThreadPoolExecutor executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize,
            this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedHandler) {

      @Override
      public void execute(Runnable command) {
        Runnable decorated = command;
        if (taskDecorator != null) {
          decorated = taskDecorator.decorate(command);
          if (decorated != command) {
            decoratedTaskMap.put(decorated, command);
          }
        }
        super.execute(decorated);
      }

      @Override
      protected void beforeExecute(Thread thread, Runnable task) {
        ThreadPoolTaskExecutor.this.beforeExecute(thread, task);
      }

      @Override
      protected void afterExecute(Runnable task, Throwable ex) {
        ThreadPoolTaskExecutor.this.afterExecute(task, ex);
      }
    };

    if (this.allowCoreThreadTimeOut) {
      executor.allowCoreThreadTimeOut(true);
    }
    if (this.prestartAllCoreThreads) {
      executor.prestartAllCoreThreads();
    }

    this.threadPoolExecutor = executor;
    return executor;
  }

  /**
   * Create the BlockingQueue to use for the ThreadPoolExecutor.
   * <p>A LinkedBlockingQueue instance will be created for a positive
   * capacity value; a SynchronousQueue else.
   *
   * @param queueCapacity the specified queue capacity
   * @return the BlockingQueue instance
   * @see java.util.concurrent.LinkedBlockingQueue
   * @see java.util.concurrent.SynchronousQueue
   */
  protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
    if (queueCapacity > 0) {
      return new LinkedBlockingQueue<>(queueCapacity);
    }
    else {
      return new SynchronousQueue<>();
    }
  }

  /**
   * Return the underlying ThreadPoolExecutor for native access.
   *
   * @return the underlying ThreadPoolExecutor (never {@code null})
   * @throws IllegalStateException if the ThreadPoolTaskExecutor hasn't been initialized yet
   */
  public ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException {
    Assert.state(this.threadPoolExecutor != null, "ThreadPoolTaskExecutor not initialized");
    return this.threadPoolExecutor;
  }

  /**
   * Return the current pool size.
   *
   * @see java.util.concurrent.ThreadPoolExecutor#getPoolSize()
   */
  public int getPoolSize() {
    if (this.threadPoolExecutor == null) {
      // Not initialized yet: assume core pool size.
      return this.corePoolSize;
    }
    return this.threadPoolExecutor.getPoolSize();
  }

  /**
   * Return the current queue size.
   *
   * @see java.util.concurrent.ThreadPoolExecutor#getQueue()
   */
  public int getQueueSize() {
    if (this.threadPoolExecutor == null) {
      // Not initialized yet: assume no queued tasks.
      return 0;
    }
    return this.threadPoolExecutor.getQueue().size();
  }

  /**
   * Return the number of currently active threads.
   *
   * @see ThreadPoolExecutor#getActiveCount()
   */
  public int getActiveCount() {
    if (this.threadPoolExecutor == null) {
      // Not initialized yet: assume no active threads.
      return 0;
    }
    return this.threadPoolExecutor.getActiveCount();
  }

  @Override
  public void execute(Runnable task) {
    Executor executor = getThreadPoolExecutor();
    try {
      executor.execute(task);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  @Override
  public Future<Void> submit(Runnable task) {
    ExecutorService executor = getThreadPoolExecutor();
    try {
      return Future.run(task, executor);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  @Override
  public <T> Future<T> submit(Callable<T> task) {
    ExecutorService executor = getThreadPoolExecutor();
    try {
      return Future.run(task, executor);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  @Override
  protected void cancelRemainingTask(Runnable task) {
    super.cancelRemainingTask(task);
    // Cancel associated user-level Future handle as well
    Object original = this.decoratedTaskMap.get(task);
    if (original instanceof java.util.concurrent.Future<?> future) {
      future.cancel(true);
    }
  }

  @Override
  protected void initiateEarlyShutdown() {
    if (this.strictEarlyShutdown) {
      super.initiateEarlyShutdown();
    }
  }
}

1	/*
2	* Copyright 2017 - 2025 the original author or authors.
3	*
4	* This program is free software: you can redistribute it and/or modify
5	* it under the terms of the GNU General Public License as published by
6	* the Free Software Foundation, either version 3 of the License, or
7	* (at your option) any later version.
8	*
9	* This program is distributed in the hope that it will be useful,
10	* but WITHOUT ANY WARRANTY; without even the implied warranty of
11	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	* GNU General Public License for more details.
13	*
14	* You should have received a copy of the GNU General Public License
15	* along with this program. If not, see [https://www.gnu.org/licenses/]
16	*/
17
18	package infra.scheduling.concurrent;
19
20	import org.jspecify.annotations.Nullable;
21
22	import java.util.concurrent.BlockingQueue;
23	import java.util.concurrent.Callable;
24	import java.util.concurrent.Executor;
25	import java.util.concurrent.ExecutorService;
26	import java.util.concurrent.LinkedBlockingQueue;
27	import java.util.concurrent.RejectedExecutionException;
28	import java.util.concurrent.RejectedExecutionHandler;
29	import java.util.concurrent.SynchronousQueue;
30	import java.util.concurrent.ThreadFactory;
31	import java.util.concurrent.ThreadPoolExecutor;
32	import java.util.concurrent.TimeUnit;
33
34	import infra.core.task.AsyncTaskExecutor;
35	import infra.core.task.TaskDecorator;
36	import infra.core.task.TaskExecutor;
37	import infra.core.task.TaskRejectedException;
38	import infra.lang.Assert;
39	import infra.scheduling.SchedulingTaskExecutor;
40	import infra.util.ConcurrentReferenceHashMap;
41	import infra.util.concurrent.Future;
42
43	/**
44	* JavaBean that allows for configuring a {@link ThreadPoolExecutor}
45	* in bean style (through its "corePoolSize", "maxPoolSize", "keepAliveSeconds", "queueCapacity"
46	* properties) and exposing it as a Infra {@link TaskExecutor}.
47	* This class is also well suited for management and monitoring (e.g. through JMX),
48	* providing several useful attributes: "corePoolSize", "maxPoolSize", "keepAliveSeconds"
49	* (all supporting updates at runtime); "poolSize", "activeCount" (for introspection only).
50	*
51	* <p>The default configuration is a core pool size of 1, with unlimited max pool size
52	* and unlimited queue capacity. This is roughly equivalent to
53	* {@link java.util.concurrent.Executors#newSingleThreadExecutor()}, sharing a single
54	* thread for all tasks. Setting {@link #setQueueCapacity "queueCapacity"} to 0 mimics
55	* {@link java.util.concurrent.Executors#newCachedThreadPool()}, with immediate scaling
56	* of threads in the pool to a potentially very high number. Consider also setting a
57	* {@link #setMaxPoolSize "maxPoolSize"} at that point, as well as possibly a higher
58	* {@link #setCorePoolSize "corePoolSize"} (see also the
59	* {@link #setAllowCoreThreadTimeOut "allowCoreThreadTimeOut"} mode of scaling).
60	*
61	* <p><b>NOTE:</b> This class implements Infra
62	* {@link TaskExecutor} interface as well as the
63	* {@link java.util.concurrent.Executor} interface, with the former being the primary
64	* interface, the other just serving as secondary convenience. For this reason, the
65	* exception handling follows the TaskExecutor contract rather than the Executor contract,
66	* in particular regarding the {@link TaskRejectedException}.
67	*
68	* <p>For an alternative, you may set up a ThreadPoolExecutor instance directly using
69	* constructor injection, or use a factory method definition that points to the
70	* {@link java.util.concurrent.Executors} class. To expose such a raw Executor as a
71	* Infra {@link TaskExecutor}, simply wrap it with a
72	* {@link ConcurrentTaskExecutor} adapter.
73	*
74	* @author Juergen Hoeller
75	* @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
76	* @see TaskExecutor
77	* @see ThreadPoolExecutor
78	* @see ThreadPoolExecutorFactoryBean
79	* @see ConcurrentTaskExecutor
80	* @since 4.0
81	*/
82	@SuppressWarnings("serial")
83	public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport implements AsyncTaskExecutor, SchedulingTaskExecutor {	2✔
84
85	private final Object poolSizeMonitor = new Object();	5✔
86
87	private int corePoolSize = 1;	3✔
88
89	private int maxPoolSize = Integer.MAX_VALUE;	3✔
90
91	private int keepAliveSeconds = 60;	3✔
92
93	private int queueCapacity = Integer.MAX_VALUE;	3✔
94
95	private boolean allowCoreThreadTimeOut = false;	3✔
96
97	private boolean prestartAllCoreThreads = false;	3✔
98
99	private boolean strictEarlyShutdown = false;	3✔
100
101	@Nullable
102	private TaskDecorator taskDecorator;
103
104	@Nullable
105	private ThreadPoolExecutor threadPoolExecutor;
106
107	// Runnable decorator to user-level FutureTask, if different
108	private final ConcurrentReferenceHashMap<Runnable, Object> decoratedTaskMap =	8✔
109	new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);
110
111	/**
112	* Set the ThreadPoolExecutor's core pool size.
113	* Default is 1.
114	* <p><b>This setting can be modified at runtime, for example through JMX.</b>
115	*/
116	public void setCorePoolSize(int corePoolSize) {
117	synchronized(this.poolSizeMonitor) {	5✔
118	if (this.threadPoolExecutor != null) {	3✔
119	this.threadPoolExecutor.setCorePoolSize(corePoolSize);	4✔
120	}
121	this.corePoolSize = corePoolSize;	3✔
122	}	3✔
123	}	1✔
124
125	/**
126	* Return the ThreadPoolExecutor's core pool size.
127	*/
128	public int getCorePoolSize() {
129	synchronized(this.poolSizeMonitor) {	5✔
130	return this.corePoolSize;	5✔
131	}
132	}
133
134	/**
135	* Set the ThreadPoolExecutor's maximum pool size.
136	* Default is {@code Integer.MAX_VALUE}.
137	* <p><b>This setting can be modified at runtime, for example through JMX.</b>
138	*/
139	public void setMaxPoolSize(int maxPoolSize) {
140	synchronized(this.poolSizeMonitor) {	5✔
141	if (this.threadPoolExecutor != null) {	3✔
142	this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);	4✔
143	}
144	this.maxPoolSize = maxPoolSize;	3✔
145	}	3✔
146	}	1✔
147
148	/**
149	* Return the ThreadPoolExecutor's maximum pool size.
150	*/
151	public int getMaxPoolSize() {
152	synchronized(this.poolSizeMonitor) {	5✔
153	return this.maxPoolSize;	5✔
154	}
155	}
156
157	/**
158	* Set the ThreadPoolExecutor's keep-alive seconds.
159	* <p>Default is 60.
160	* <p><b>This setting can be modified at runtime, for example through JMX.</b>
161	*/
162	public void setKeepAliveSeconds(int keepAliveSeconds) {
163	synchronized(this.poolSizeMonitor) {	5✔
164	if (this.threadPoolExecutor != null) {	3✔
165	this.threadPoolExecutor.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);	6✔
166	}
167	this.keepAliveSeconds = keepAliveSeconds;	3✔
168	}	3✔
169	}	1✔
170
171	/**
172	* Return the ThreadPoolExecutor's keep-alive seconds.
173	*/
174	public int getKeepAliveSeconds() {
175	synchronized(this.poolSizeMonitor) {	5✔
176	return this.keepAliveSeconds;	5✔
177	}
178	}
179
180	/**
181	* Set the capacity for the ThreadPoolExecutor's BlockingQueue.
182	* <p>Default is {@code Integer.MAX_VALUE}.
183	* <p>Any positive value will lead to a LinkedBlockingQueue instance;
184	* any other value will lead to a SynchronousQueue instance.
185	*
186	* @see LinkedBlockingQueue
187	* @see SynchronousQueue
188	*/
189	public void setQueueCapacity(int queueCapacity) {
190	this.queueCapacity = queueCapacity;	3✔
191	}	1✔
192
193	/**
194	* Return the capacity for the ThreadPoolExecutor's BlockingQueue.
195	*
196	* @see #setQueueCapacity(int)
197	*/
198	public int getQueueCapacity() {
199	return this.queueCapacity;	3✔
200	}
201
202	/**
203	* Specify whether to allow core threads to time out. This enables dynamic
204	* growing and shrinking even in combination with a non-zero queue (since
205	* the max pool size will only grow once the queue is full).
206	* <p>Default is "false".
207	*
208	* @see ThreadPoolExecutor#allowCoreThreadTimeOut(boolean)
209	*/
210	public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
211	this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;	3✔
212	}	1✔
213
214	/**
215	* Specify whether to start all core threads, causing them to idly wait for work.
216	* <p>Default is "false", starting threads and adding them to the pool on demand.
217	*
218	* @see ThreadPoolExecutor#prestartAllCoreThreads
219	*/
220	public void setPrestartAllCoreThreads(boolean prestartAllCoreThreads) {
221	this.prestartAllCoreThreads = prestartAllCoreThreads;	×
222	}	×
223
224	/**
225	* Specify whether to initiate an early shutdown signal on context close,
226	* disposing all idle threads and rejecting further task submissions.
227	* <p>By default, existing tasks will be allowed to complete within the
228	* coordinated lifecycle stop phase in any case. This setting just controls
229	* whether an explicit {@link ThreadPoolExecutor#shutdown()} call will be
230	* triggered on context close, rejecting task submissions after that point.
231	* <p>the default is "false", leniently allowing for late tasks
232	* to arrive after context close, still participating in the lifecycle stop
233	* phase. Note that this differs from {@link #setAcceptTasksAfterContextClose}
234	* which completely bypasses the coordinated lifecycle stop phase, with no
235	* explicit waiting for the completion of existing tasks at all.
236	* <p>Switch this to "true" for a strict early shutdown signal analogous to
237	* the 4.0-established default behavior of {@link ThreadPoolTaskScheduler}.
238	* Note that the related flags {@link #setAcceptTasksAfterContextClose} and
239	* {@link #setWaitForTasksToCompleteOnShutdown} will override this setting,
240	* leading to a late shutdown without a coordinated lifecycle stop phase.
241	*
242	* @see #initiateShutdown()
243	*/
244	public void setStrictEarlyShutdown(boolean defaultEarlyShutdown) {
245	this.strictEarlyShutdown = defaultEarlyShutdown;	×
246	}	×
247
248	/**
249	* Specify a custom {@link TaskDecorator} to be applied to any {@link Runnable}
250	* about to be executed.
251	* <p>Note that such a decorator is not necessarily being applied to the
252	* user-supplied {@code Runnable}/{@code Callable} but rather to the actual
253	* execution callback (which may be a wrapper around the user-supplied task).
254	* <p>The primary use case is to set some execution context around the task's
255	* invocation, or to provide some monitoring/statistics for task execution.
256	* <p><b>NOTE:</b> Exception handling in {@code TaskDecorator} implementations
257	* is limited to plain {@code Runnable} execution via {@code execute} calls.
258	* In case of {@code #submit} calls, the exposed {@code Runnable} will be a
259	* {@code FutureTask} which does not propagate any exceptions; you might
260	* have to cast it and call {@code Future#get} to evaluate exceptions.
261	* See the {@code ThreadPoolExecutor#afterExecute} javadoc for an example
262	* of how to access exceptions in such a {@code Future} case.
263	*/
264	public void setTaskDecorator(TaskDecorator taskDecorator) {
265	this.taskDecorator = taskDecorator;	3✔
266	}	1✔
267
268	/**
269	* Note: This method exposes an {@link ExecutorService} to its base class
270	* but stores the actual {@link ThreadPoolExecutor} handle internally.
271	* Do not override this method for replacing the executor, rather just for
272	* decorating its {@code ExecutorService} handle or storing custom state.
273	*/
274	@Override
275	protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedHandler) {
276	BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);	5✔
277
278	ThreadPoolExecutor executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize,	16✔
279	this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedHandler) {	13✔
280
281	@Override
282	public void execute(Runnable command) {
283	Runnable decorated = command;	2✔
284	if (taskDecorator != null) {	4✔
285	decorated = taskDecorator.decorate(command);	6✔
286	if (decorated != command) {	3!
287	decoratedTaskMap.put(decorated, command);	7✔
288	}
289	}
290	super.execute(decorated);	3✔
291	}	1✔
292
293	@Override
294	protected void beforeExecute(Thread thread, Runnable task) {
295	ThreadPoolTaskExecutor.this.beforeExecute(thread, task);	5✔
296	}	1✔
297
298	@Override
299	protected void afterExecute(Runnable task, Throwable ex) {
300	ThreadPoolTaskExecutor.this.afterExecute(task, ex);	5✔
301	}	1✔
302	};
303
304	if (this.allowCoreThreadTimeOut) {	3✔
305	executor.allowCoreThreadTimeOut(true);	3✔
306	}
307	if (this.prestartAllCoreThreads) {	3!
308	executor.prestartAllCoreThreads();	×
309	}
310
311	this.threadPoolExecutor = executor;	3✔
312	return executor;	2✔
313	}
314
315	/**
316	* Create the BlockingQueue to use for the ThreadPoolExecutor.
317	* <p>A LinkedBlockingQueue instance will be created for a positive
318	* capacity value; a SynchronousQueue else.
319	*
320	* @param queueCapacity the specified queue capacity
321	* @return the BlockingQueue instance
322	* @see java.util.concurrent.LinkedBlockingQueue
323	* @see java.util.concurrent.SynchronousQueue
324	*/
325	protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
326	if (queueCapacity > 0) {	2✔
327	return new LinkedBlockingQueue<>(queueCapacity);	5✔
328	}
329	else {
330	return new SynchronousQueue<>();	4✔
331	}
332	}
333
334	/**
335	* Return the underlying ThreadPoolExecutor for native access.
336	*
337	* @return the underlying ThreadPoolExecutor (never {@code null})
338	* @throws IllegalStateException if the ThreadPoolTaskExecutor hasn't been initialized yet
339	*/
340	public ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException {
341	Assert.state(this.threadPoolExecutor != null, "ThreadPoolTaskExecutor not initialized");	8✔
342	return this.threadPoolExecutor;	3✔
343	}
344
345	/**
346	* Return the current pool size.
347	*
348	* @see java.util.concurrent.ThreadPoolExecutor#getPoolSize()
349	*/
350	public int getPoolSize() {
351	if (this.threadPoolExecutor == null) {	×
352	// Not initialized yet: assume core pool size.
353	return this.corePoolSize;	×
354	}
355	return this.threadPoolExecutor.getPoolSize();	×
356	}
357
358	/**
359	* Return the current queue size.
360	*
361	* @see java.util.concurrent.ThreadPoolExecutor#getQueue()
362	*/
363	public int getQueueSize() {
364	if (this.threadPoolExecutor == null) {	3✔
365	// Not initialized yet: assume no queued tasks.
366	return 0;	2✔
367	}
368	return this.threadPoolExecutor.getQueue().size();	5✔
369	}
370
371	/**
372	* Return the number of currently active threads.
373	*
374	* @see ThreadPoolExecutor#getActiveCount()
375	*/
376	public int getActiveCount() {
377	if (this.threadPoolExecutor == null) {	×
378	// Not initialized yet: assume no active threads.
379	return 0;	×
380	}
381	return this.threadPoolExecutor.getActiveCount();	×
382	}
383
384	@Override
385	public void execute(Runnable task) {
386	Executor executor = getThreadPoolExecutor();	3✔
387	try {
388	executor.execute(task);	3✔
389	}
390	catch (RejectedExecutionException ex) {	×
391	throw new TaskRejectedException(executor, task, ex);	×
392	}	1✔
393	}	1✔
394
395	@Override
396	public Future<Void> submit(Runnable task) {
397	ExecutorService executor = getThreadPoolExecutor();	3✔
398	try {
399	return Future.run(task, executor);	4✔
400	}
401	catch (RejectedExecutionException ex) {	×
402	throw new TaskRejectedException(executor, task, ex);	×
403	}
404	}
405
406	@Override
407	public <T> Future<T> submit(Callable<T> task) {
408	ExecutorService executor = getThreadPoolExecutor();	3✔
409	try {
410	return Future.run(task, executor);	4✔
411	}
412	catch (RejectedExecutionException ex) {	×
413	throw new TaskRejectedException(executor, task, ex);	×
414	}
415	}
416
417	@Override
418	protected void cancelRemainingTask(Runnable task) {
419	super.cancelRemainingTask(task);	3✔
420	// Cancel associated user-level Future handle as well
421	Object original = this.decoratedTaskMap.get(task);	5✔
422	if (original instanceof java.util.concurrent.Future<?> future) {	6✔
423	future.cancel(true);	4✔
424	}
425	}	1✔
426
427	@Override
428	protected void initiateEarlyShutdown() {
429	if (this.strictEarlyShutdown) {	3!
430	super.initiateEarlyShutdown();	×
431	}
432	}	1✔
433	}

TAKETODAY / today-infrastructure / 18154768944

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous