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

91.8

today-context/src/main/java/infra/scheduling/concurrent/ThreadPoolExecutorFactoryBean.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.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
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.beans.factory.FactoryBean;

/**
 * JavaBean that allows for configuring a {@link ThreadPoolExecutor}
 * in bean style (through its "corePoolSize", "maxPoolSize", "keepAliveSeconds",
 * "queueCapacity" properties) and exposing it as a bean reference of its native
 * {@link ExecutorService} type.
 *
 * <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 Executors#newSingleThreadExecutor()}, sharing a single
 * thread for all tasks. Setting {@link #setQueueCapacity "queueCapacity"} to 0 mimics
 * {@link 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>For an alternative, you may set up a {@link ThreadPoolExecutor} instance directly
 * using constructor injection, or use a factory method definition that points to the
 * {@link Executors} class.
 * <b>This is strongly recommended in particular for common {@code @Component} methods in
 * configuration classes, where this {@code FactoryBean} variant would force you to
 * return the {@code FactoryBean} type instead of the actual {@code Executor} type.</b>
 *
 * <p>If you need a timing-based {@link java.util.concurrent.ScheduledExecutorService}
 * instead, consider {@link ScheduledExecutorFactoryBean}.
 *
 * @author Juergen Hoeller
 * @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
 * @see ExecutorService
 * @see Executors
 * @see ThreadPoolExecutor
 * @since 4.0
 */
@SuppressWarnings("serial")
public class ThreadPoolExecutorFactoryBean extends ExecutorConfigurationSupport
        implements FactoryBean<ExecutorService> {

  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;

  private boolean exposeUnconfigurableExecutor = false;

  @Nullable
  private ExecutorService exposedExecutor;

  /**
   * Set the ThreadPoolExecutor's core pool size.
   * Default is 1.
   */
  public void setCorePoolSize(int corePoolSize) {
    this.corePoolSize = corePoolSize;
  }

  /**
   * Set the ThreadPoolExecutor's maximum pool size.
   * Default is {@code Integer.MAX_VALUE}.
   */
  public void setMaxPoolSize(int maxPoolSize) {
    this.maxPoolSize = maxPoolSize;
  }

  /**
   * Set the ThreadPoolExecutor's keep-alive seconds.
   * Default is 60.
   */
  public void setKeepAliveSeconds(int keepAliveSeconds) {
    this.keepAliveSeconds = keepAliveSeconds;
  }

  /**
   * Set the capacity for the ThreadPoolExecutor's BlockingQueue.
   * 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 java.util.concurrent.LinkedBlockingQueue
   * @see java.util.concurrent.SynchronousQueue
   */
  public void setQueueCapacity(int queueCapacity) {
    this.queueCapacity = 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 java.util.concurrent.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".
   *
   * @see java.util.concurrent.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>Default is "false".
   * See {@link ThreadPoolTaskExecutor#setStrictEarlyShutdown} for details.
   *
   * @see #initiateShutdown()
   */
  public void setStrictEarlyShutdown(boolean defaultEarlyShutdown) {
    this.strictEarlyShutdown = defaultEarlyShutdown;
  }

  /**
   * Specify whether this FactoryBean should expose an unconfigurable
   * decorator for the created executor.
   * <p>Default is "false", exposing the raw executor as bean reference.
   * Switch this flag to "true" to strictly prevent clients from
   * modifying the executor's configuration.
   *
   * @see java.util.concurrent.Executors#unconfigurableExecutorService
   */
  public void setExposeUnconfigurableExecutor(boolean exposeUnconfigurableExecutor) {
    this.exposeUnconfigurableExecutor = exposeUnconfigurableExecutor;
  }

  @Override
  protected ExecutorService initializeExecutor(
          ThreadFactory threadFactory, RejectedExecutionHandler rejectedHandler) {

    BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
    ThreadPoolExecutor executor = createExecutor(this.corePoolSize, this.maxPoolSize,
            this.keepAliveSeconds, queue, threadFactory, rejectedHandler);
    if (this.allowCoreThreadTimeOut) {
      executor.allowCoreThreadTimeOut(true);
    }
    if (this.prestartAllCoreThreads) {
      executor.prestartAllCoreThreads();
    }

    // Wrap executor with an unconfigurable decorator.
    this.exposedExecutor = (this.exposeUnconfigurableExecutor ?
            Executors.unconfigurableExecutorService(executor) : executor);

    return executor;
  }

  /**
   * Create a new instance of {@link ThreadPoolExecutor} or a subclass thereof.
   * <p>The default implementation creates a standard {@link ThreadPoolExecutor}.
   * Can be overridden to provide custom {@link ThreadPoolExecutor} subclasses.
   *
   * @param corePoolSize the specified core pool size
   * @param maxPoolSize the specified maximum pool size
   * @param keepAliveSeconds the specified keep-alive time in seconds
   * @param queue the BlockingQueue to use
   * @param threadFactory the ThreadFactory to use
   * @param rejectedExecutionHandler the RejectedExecutionHandler to use
   * @return a new ThreadPoolExecutor instance
   * @see #afterPropertiesSet()
   */
  protected ThreadPoolExecutor createExecutor(
          int corePoolSize, int maxPoolSize, int keepAliveSeconds, BlockingQueue<Runnable> queue,
          ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

    return new ThreadPoolExecutor(corePoolSize, maxPoolSize,
            keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler) {
      @Override
      protected void beforeExecute(Thread thread, Runnable task) {
        ThreadPoolExecutorFactoryBean.this.beforeExecute(thread, task);
      }

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

  /**
   * 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<>();
    }
  }

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

  @Override
  @Nullable
  public ExecutorService getObject() {
    return this.exposedExecutor;
  }

  @Override
  public Class<? extends ExecutorService> getObjectType() {
    return (this.exposedExecutor != null ? this.exposedExecutor.getClass() : ExecutorService.class);
  }

  @Override
  public boolean isSingleton() {
    return true;
  }

}

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.ExecutorService;
24	import java.util.concurrent.Executors;
25	import java.util.concurrent.LinkedBlockingQueue;
26	import java.util.concurrent.RejectedExecutionHandler;
27	import java.util.concurrent.SynchronousQueue;
28	import java.util.concurrent.ThreadFactory;
29	import java.util.concurrent.ThreadPoolExecutor;
30	import java.util.concurrent.TimeUnit;
31
32	import infra.beans.factory.FactoryBean;
33
34	/**
35	* JavaBean that allows for configuring a {@link ThreadPoolExecutor}
36	* in bean style (through its "corePoolSize", "maxPoolSize", "keepAliveSeconds",
37	* "queueCapacity" properties) and exposing it as a bean reference of its native
38	* {@link ExecutorService} type.
39	*
40	* <p>The default configuration is a core pool size of 1, with unlimited max pool size
41	* and unlimited queue capacity. This is roughly equivalent to
42	* {@link Executors#newSingleThreadExecutor()}, sharing a single
43	* thread for all tasks. Setting {@link #setQueueCapacity "queueCapacity"} to 0 mimics
44	* {@link Executors#newCachedThreadPool()}, with immediate scaling
45	* of threads in the pool to a potentially very high number. Consider also setting a
46	* {@link #setMaxPoolSize "maxPoolSize"} at that point, as well as possibly a higher
47	* {@link #setCorePoolSize "corePoolSize"} (see also the
48	* {@link #setAllowCoreThreadTimeOut "allowCoreThreadTimeOut"} mode of scaling).
49	*
50	* <p>For an alternative, you may set up a {@link ThreadPoolExecutor} instance directly
51	* using constructor injection, or use a factory method definition that points to the
52	* {@link Executors} class.
53	* <b>This is strongly recommended in particular for common {@code @Component} methods in
54	* configuration classes, where this {@code FactoryBean} variant would force you to
55	* return the {@code FactoryBean} type instead of the actual {@code Executor} type.</b>
56	*
57	* <p>If you need a timing-based {@link java.util.concurrent.ScheduledExecutorService}
58	* instead, consider {@link ScheduledExecutorFactoryBean}.
59	*
60	* @author Juergen Hoeller
61	* @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
62	* @see ExecutorService
63	* @see Executors
64	* @see ThreadPoolExecutor
65	* @since 4.0
66	*/
67	@SuppressWarnings("serial")
68	public class ThreadPoolExecutorFactoryBean extends ExecutorConfigurationSupport	2✔
69	implements FactoryBean<ExecutorService> {
70
71	private int corePoolSize = 1;	3✔
72
73	private int maxPoolSize = Integer.MAX_VALUE;	3✔
74
75	private int keepAliveSeconds = 60;	3✔
76
77	private int queueCapacity = Integer.MAX_VALUE;	3✔
78
79	private boolean allowCoreThreadTimeOut = false;	3✔
80
81	private boolean prestartAllCoreThreads = false;	3✔
82
83	private boolean strictEarlyShutdown = false;	3✔
84
85	private boolean exposeUnconfigurableExecutor = false;	4✔
86
87	@Nullable
88	private ExecutorService exposedExecutor;
89
90	/**
91	* Set the ThreadPoolExecutor's core pool size.
92	* Default is 1.
93	*/
94	public void setCorePoolSize(int corePoolSize) {
95	this.corePoolSize = corePoolSize;	3✔
96	}	1✔
97
98	/**
99	* Set the ThreadPoolExecutor's maximum pool size.
100	* Default is {@code Integer.MAX_VALUE}.
101	*/
102	public void setMaxPoolSize(int maxPoolSize) {
103	this.maxPoolSize = maxPoolSize;	3✔
104	}	1✔
105
106	/**
107	* Set the ThreadPoolExecutor's keep-alive seconds.
108	* Default is 60.
109	*/
110	public void setKeepAliveSeconds(int keepAliveSeconds) {
111	this.keepAliveSeconds = keepAliveSeconds;	3✔
112	}	1✔
113
114	/**
115	* Set the capacity for the ThreadPoolExecutor's BlockingQueue.
116	* Default is {@code Integer.MAX_VALUE}.
117	* <p>Any positive value will lead to a LinkedBlockingQueue instance;
118	* any other value will lead to a SynchronousQueue instance.
119	*
120	* @see java.util.concurrent.LinkedBlockingQueue
121	* @see java.util.concurrent.SynchronousQueue
122	*/
123	public void setQueueCapacity(int queueCapacity) {
124	this.queueCapacity = queueCapacity;	3✔
125	}	1✔
126
127	/**
128	* Specify whether to allow core threads to time out. This enables dynamic
129	* growing and shrinking even in combination with a non-zero queue (since
130	* the max pool size will only grow once the queue is full).
131	* <p>Default is "false".
132	*
133	* @see java.util.concurrent.ThreadPoolExecutor#allowCoreThreadTimeOut(boolean)
134	*/
135	public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
136	this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;	3✔
137	}	1✔
138
139	/**
140	* Specify whether to start all core threads, causing them to idly wait for work.
141	* <p>Default is "false".
142	*
143	* @see java.util.concurrent.ThreadPoolExecutor#prestartAllCoreThreads
144	*/
145	public void setPrestartAllCoreThreads(boolean prestartAllCoreThreads) {
146	this.prestartAllCoreThreads = prestartAllCoreThreads;	3✔
147	}	1✔
148
149	/**
150	* Specify whether to initiate an early shutdown signal on context close,
151	* disposing all idle threads and rejecting further task submissions.
152	* <p>Default is "false".
153	* See {@link ThreadPoolTaskExecutor#setStrictEarlyShutdown} for details.
154	*
155	* @see #initiateShutdown()
156	*/
157	public void setStrictEarlyShutdown(boolean defaultEarlyShutdown) {
158	this.strictEarlyShutdown = defaultEarlyShutdown;	3✔
159	}	1✔
160
161	/**
162	* Specify whether this FactoryBean should expose an unconfigurable
163	* decorator for the created executor.
164	* <p>Default is "false", exposing the raw executor as bean reference.
165	* Switch this flag to "true" to strictly prevent clients from
166	* modifying the executor's configuration.
167	*
168	* @see java.util.concurrent.Executors#unconfigurableExecutorService
169	*/
170	public void setExposeUnconfigurableExecutor(boolean exposeUnconfigurableExecutor) {
171	this.exposeUnconfigurableExecutor = exposeUnconfigurableExecutor;	×
172	}	×
173
174	@Override
175	protected ExecutorService initializeExecutor(
176	ThreadFactory threadFactory, RejectedExecutionHandler rejectedHandler) {
177
178	BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);	5✔
179	ThreadPoolExecutor executor = createExecutor(this.corePoolSize, this.maxPoolSize,	12✔
180	this.keepAliveSeconds, queue, threadFactory, rejectedHandler);
181	if (this.allowCoreThreadTimeOut) {	3✔
182	executor.allowCoreThreadTimeOut(true);	3✔
183	}
184	if (this.prestartAllCoreThreads) {	3✔
185	executor.prestartAllCoreThreads();	3✔
186	}
187
188	// Wrap executor with an unconfigurable decorator.
189	this.exposedExecutor = (this.exposeUnconfigurableExecutor ?	4!
190	Executors.unconfigurableExecutorService(executor) : executor);	2✔
191
192	return executor;	2✔
193	}
194
195	/**
196	* Create a new instance of {@link ThreadPoolExecutor} or a subclass thereof.
197	* <p>The default implementation creates a standard {@link ThreadPoolExecutor}.
198	* Can be overridden to provide custom {@link ThreadPoolExecutor} subclasses.
199	*
200	* @param corePoolSize the specified core pool size
201	* @param maxPoolSize the specified maximum pool size
202	* @param keepAliveSeconds the specified keep-alive time in seconds
203	* @param queue the BlockingQueue to use
204	* @param threadFactory the ThreadFactory to use
205	* @param rejectedExecutionHandler the RejectedExecutionHandler to use
206	* @return a new ThreadPoolExecutor instance
207	* @see #afterPropertiesSet()
208	*/
209	protected ThreadPoolExecutor createExecutor(
210	int corePoolSize, int maxPoolSize, int keepAliveSeconds, BlockingQueue<Runnable> queue,
211	ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
212
213	return new ThreadPoolExecutor(corePoolSize, maxPoolSize,	13✔
214	keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler) {	13✔
215	@Override
216	protected void beforeExecute(Thread thread, Runnable task) {
217	ThreadPoolExecutorFactoryBean.this.beforeExecute(thread, task);	5✔
218	}	1✔
219
220	@Override
221	protected void afterExecute(Runnable task, Throwable ex) {
222	ThreadPoolExecutorFactoryBean.this.afterExecute(task, ex);	5✔
223	}	1✔
224	};
225	}
226
227	/**
228	* Create the BlockingQueue to use for the ThreadPoolExecutor.
229	* <p>A LinkedBlockingQueue instance will be created for a positive
230	* capacity value; a SynchronousQueue else.
231	*
232	* @param queueCapacity the specified queue capacity
233	* @return the BlockingQueue instance
234	* @see java.util.concurrent.LinkedBlockingQueue
235	* @see java.util.concurrent.SynchronousQueue
236	*/
237	protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
238	if (queueCapacity > 0) {	2✔
239	return new LinkedBlockingQueue<>(queueCapacity);	5✔
240	}
241	else {
242	return new SynchronousQueue<>();	4✔
243	}
244	}
245
246	@Override
247	protected void initiateEarlyShutdown() {
248	if (this.strictEarlyShutdown) {	3!
249	super.initiateEarlyShutdown();	×
250	}
251	}	1✔
252
253	@Override
254	@Nullable
255	public ExecutorService getObject() {
256	return this.exposedExecutor;	3✔
257	}
258
259	@Override
260	public Class<? extends ExecutorService> getObjectType() {
261	return (this.exposedExecutor != null ? this.exposedExecutor.getClass() : ExecutorService.class);	9✔
262	}
263
264	@Override
265	public boolean isSingleton() {
266	return true;	2✔
267	}
268
269	}

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