KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > springframework > scheduling > backportconcurrent > ThreadPoolTaskExecutor


1 /*
2  * Copyright 2002-2007 the original author or authors.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16
17 package org.springframework.scheduling.backportconcurrent;
18
19 import edu.emory.mathcs.backport.java.util.concurrent.BlockingQueue;
20 import edu.emory.mathcs.backport.java.util.concurrent.Executor;
21 import edu.emory.mathcs.backport.java.util.concurrent.Executors;
22 import edu.emory.mathcs.backport.java.util.concurrent.LinkedBlockingQueue;
23 import edu.emory.mathcs.backport.java.util.concurrent.RejectedExecutionException;
24 import edu.emory.mathcs.backport.java.util.concurrent.RejectedExecutionHandler;
25 import edu.emory.mathcs.backport.java.util.concurrent.SynchronousQueue;
26 import edu.emory.mathcs.backport.java.util.concurrent.ThreadFactory;
27 import edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor;
28 import edu.emory.mathcs.backport.java.util.concurrent.TimeUnit;
29 import org.apache.commons.logging.Log;
30 import org.apache.commons.logging.LogFactory;
31
32 import org.springframework.beans.factory.DisposableBean;
33 import org.springframework.beans.factory.InitializingBean;
34 import org.springframework.core.task.TaskRejectedException;
35 import org.springframework.scheduling.SchedulingTaskExecutor;
36 import org.springframework.util.Assert;
37
38 /**
39  * JavaBean that allows for configuring a JSR-166 backport
40  * {@link edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor} in bean
41  * style (through its "corePoolSize", "maxPoolSize", "keepAliveSeconds", "queueCapacity"
42  * properties), exposing it as a Spring {@link org.springframework.core.task.TaskExecutor}.
43  * This is an alternative to configuring a ThreadPoolExecutor instance directly using
44  * constructor injection, with a separate {@link ConcurrentTaskExecutor} adapter wrapping it.
45  *
46  * <p>For any custom needs, in particular for defining a
47  * {@link edu.emory.mathcs.backport.java.util.concurrent.ScheduledThreadPoolExecutor},
48  * it is recommended to use a straight definition of the Executor instance or a
49  * factory method definition that points to the JSR-166 backport
50  * {@link edu.emory.mathcs.backport.java.util.concurrent.Executors} class.
51  * To expose such a raw Executor as a Spring {@link org.springframework.core.task.TaskExecutor},
52  * simply wrap it with a {@link ConcurrentTaskExecutor} adapter.
53  *
54  * <p><b>NOTE:</b> This class implements Spring's
55  * {@link org.springframework.core.task.TaskExecutor} interface as well as
56  * the JSR-166 {@link edu.emory.mathcs.backport.java.util.concurrent.Executor}
57  * interface, with the former being the primary interface, the other just
58  * serving as secondary convenience. For this reason, the exception handling
59  * follows the TaskExecutor contract rather than the Executor contract, in
60  * particular regarding the {@link org.springframework.core.task.TaskRejectedException}.
61  *
62  * @author Juergen Hoeller
63  * @since 2.0.3
64  * @see org.springframework.core.task.TaskExecutor
65  * @see edu.emory.mathcs.backport.java.util.concurrent.Executor
66  * @see edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor
67  * @see edu.emory.mathcs.backport.java.util.concurrent.ScheduledThreadPoolExecutor
68  * @see edu.emory.mathcs.backport.java.util.concurrent.Executors
69  * @see ConcurrentTaskExecutor
70  */
71 public class ThreadPoolTaskExecutor implements SchedulingTaskExecutor, Executor, InitializingBean, DisposableBean {
72
73     protected final Log logger = LogFactory.getLog(getClass());
74
75     private final Object JavaDoc poolSizeMonitor = new Object JavaDoc();
76
77     private int corePoolSize = 1;
78
79     private int maxPoolSize = Integer.MAX_VALUE;
80
81     private int keepAliveSeconds = 60;
82
83     private int queueCapacity = Integer.MAX_VALUE;
84
85     private ThreadFactory threadFactory = Executors.defaultThreadFactory();
86
87     private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();
88
89     private ThreadPoolExecutor threadPoolExecutor;
90
91
92     /**
93      * Set the ThreadPoolExecutor's core pool size.
94      * Default is 1.
95      * <p><b>This setting can be modified at runtime, for example through JMX.</b>
96      */
97     public void setCorePoolSize(int corePoolSize) {
98         synchronized (this.poolSizeMonitor) {
99             this.corePoolSize = corePoolSize;
100             if (this.threadPoolExecutor != null) {
101                 this.threadPoolExecutor.setCorePoolSize(corePoolSize);
102             }
103         }
104     }
105
106     /**
107      * Return the ThreadPoolExecutor's core pool size.
108      */
109     public int getCorePoolSize() {
110         synchronized (this.poolSizeMonitor) {
111             return this.corePoolSize;
112         }
113     }
114
115     /**
116      * Set the ThreadPoolExecutor's maximum pool size.
117      * Default is <code>Integer.MAX_VALUE</code>.
118      * <p><b>This setting can be modified at runtime, for example through JMX.</b>
119      */
120     public void setMaxPoolSize(int maxPoolSize) {
121         synchronized (this.poolSizeMonitor) {
122             this.maxPoolSize = maxPoolSize;
123             if (this.threadPoolExecutor != null) {
124                 this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);
125             }
126         }
127     }
128
129     /**
130      * Return the ThreadPoolExecutor's maximum pool size.
131      */
132     public int getMaxPoolSize() {
133         synchronized (this.poolSizeMonitor) {
134             return this.maxPoolSize;
135         }
136     }
137
138     /**
139      * Set the ThreadPoolExecutor's keep-alive seconds.
140      * Default is 60.
141      * <p><b>This setting can be modified at runtime, for example through JMX.</b>
142      */
143     public void setKeepAliveSeconds(int keepAliveSeconds) {
144         synchronized (this.poolSizeMonitor) {
145             this.keepAliveSeconds = keepAliveSeconds;
146             if (this.threadPoolExecutor != null) {
147                 this.threadPoolExecutor.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);
148             }
149         }
150     }
151
152     /**
153      * Return the ThreadPoolExecutor's keep-alive seconds.
154      */
155     public int getKeepAliveSeconds() {
156         synchronized (this.poolSizeMonitor) {
157             return this.keepAliveSeconds;
158         }
159     }
160
161     /**
162      * Set the capacity for the ThreadPoolExecutor's BlockingQueue.
163      * Default is <code>Integer.MAX_VALUE</code>.
164      * <p>Any positive value will lead to a LinkedBlockingQueue instance;
165      * any other value will lead to a SynchronousQueue instance.
166      * @see edu.emory.mathcs.backport.java.util.concurrent.LinkedBlockingQueue
167      * @see edu.emory.mathcs.backport.java.util.concurrent.SynchronousQueue
168      */
169     public void setQueueCapacity(int queueCapacity) {
170         this.queueCapacity = queueCapacity;
171     }
172
173     /**
174      * Set the ThreadFactory to use for the ThreadPoolExecutor's thread pool.
175      * Default is the ThreadPoolExecutor's default thread factory.
176      * @see edu.emory.mathcs.backport.java.util.concurrent.Executors#defaultThreadFactory()
177      */
178     public void setThreadFactory(ThreadFactory threadFactory) {
179         this.threadFactory = (threadFactory != null ? threadFactory : Executors.defaultThreadFactory());
180     }
181
182     /**
183      * Set the RejectedExecutionHandler to use for the ThreadPoolExecutor.
184      * Default is the ThreadPoolExecutor's default abort policy.
185      * @see edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor.AbortPolicy
186      */
187     public void setRejectedExecutionHandler(RejectedExecutionHandler rejectedExecutionHandler) {
188         this.rejectedExecutionHandler =
189                 (rejectedExecutionHandler != null ? rejectedExecutionHandler : new ThreadPoolExecutor.AbortPolicy());
190     }
191
192
193     /**
194      * Calls <code>initialize()</code> after the container applied all property values.
195      * @see #initialize()
196      */
197     public void afterPropertiesSet() {
198         initialize();
199     }
200
201     /**
202      * Creates the BlockingQueue and the ThreadPoolExecutor.
203      * @see #createQueue
204      */
205     public void initialize() {
206         logger.info("Creating ThreadPoolExecutor");
207         BlockingQueue queue = createQueue(this.queueCapacity);
208         this.threadPoolExecutor = new ThreadPoolExecutor(
209                 this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
210                 queue, this.threadFactory, this.rejectedExecutionHandler);
211     }
212
213     /**
214      * Create the BlockingQueue to use for the ThreadPoolExecutor.
215      * <p>A LinkedBlockingQueue instance will be created for a positive
216      * capacity value; a SynchronousQueue else.
217      * @param queueCapacity the specified queue capacity
218      * @return the BlockingQueue instance
219      * @see edu.emory.mathcs.backport.java.util.concurrent.LinkedBlockingQueue
220      * @see edu.emory.mathcs.backport.java.util.concurrent.SynchronousQueue
221      */
222     protected BlockingQueue createQueue(int queueCapacity) {
223         if (queueCapacity > 0) {
224             return new LinkedBlockingQueue(queueCapacity);
225         }
226         else {
227             return new SynchronousQueue();
228         }
229     }
230
231     /**
232      * Return the underlying ThreadPoolExecutor for native access.
233      * @return the underlying ThreadPoolExecutor (never <code>null</code>)
234      * @throws IllegalStateException if the ThreadPoolTaskExecutor hasn't been initialized yet
235      */
236     public ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException JavaDoc {
237         Assert.state(this.threadPoolExecutor != null, "ThreadPoolTaskExecutor not initialized");
238         return this.threadPoolExecutor;
239     }
240
241
242     /**
243      * Implementation of both the JSR-166 backport Executor interface and the Spring
244      * TaskExecutor interface, delegating to the ThreadPoolExecutor instance.
245      * @see edu.emory.mathcs.backport.java.util.concurrent.Executor#execute(Runnable)
246      * @see org.springframework.core.task.TaskExecutor#execute(Runnable)
247      */
248     public void execute(Runnable JavaDoc task) {
249         Executor executor = getThreadPoolExecutor();
250         try {
251             executor.execute(task);
252         }
253         catch (RejectedExecutionException ex) {
254             throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
255         }
256     }
257
258     /**
259      * This task executor prefers short-lived work units.
260      */
261     public boolean prefersShortLivedTasks() {
262         return true;
263     }
264
265
266     /**
267      * Return the current pool size.
268      * @see edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor#getPoolSize()
269      */
270     public int getPoolSize() {
271         return getThreadPoolExecutor().getPoolSize();
272     }
273
274     /**
275      * Return the number of currently active threads.
276      * @see edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor#getActiveCount()
277      */
278     public int getActiveCount() {
279         return getThreadPoolExecutor().getActiveCount();
280     }
281
282
283     /**
284      * Calls <code>shutdown</code> when the BeanFactory destroys
285      * the task executor instance.
286      * @see #shutdown()
287      */
288     public void destroy() {
289         shutdown();
290     }
291
292     /**
293      * Perform a shutdown on the ThreadPoolExecutor.
294      * @see edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor#shutdown()
295      */
296     public void shutdown() {
297         logger.info("Shutting down ThreadPoolExecutor");
298         this.threadPoolExecutor.shutdown();
299     }
300
301 }
302
Popular Tags