1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
|
//
// ========================================================================
// Copyright (c) 1995-2016 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
//
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
//
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
//
package org.eclipse.jetty.util.thread;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.util.component.AbstractLifeCycle;
import org.eclipse.jetty.util.component.LifeCycle;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
/* ------------------------------------------------------------ */
/**
* Jetty ThreadPool using java 5 ThreadPoolExecutor
* This class wraps a {@link ExecutorService} as a {@link ThreadPool} and
* {@link LifeCycle} interfaces so that it may be used by the Jetty <code>org.eclipse.jetty.server.Server</code>
*/
public class ExecutorThreadPool extends AbstractLifeCycle implements ThreadPool, LifeCycle
{
private static final Logger LOG = Log.getLogger(ExecutorThreadPool.class);
private final ExecutorService _executor;
/* ------------------------------------------------------------ */
public ExecutorThreadPool(ExecutorService executor)
{
_executor = executor;
}
/* ------------------------------------------------------------ */
/**
* Wraps an {@link ThreadPoolExecutor}.
* Max pool size is 256, pool thread timeout after 60 seconds and
* an unbounded {@link LinkedBlockingQueue} is used for the job queue;
*/
public ExecutorThreadPool()
{
// Using an unbounded queue makes the maxThreads parameter useless
// Refer to ThreadPoolExecutor javadocs for details
this(new ThreadPoolExecutor(256, 256, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()));
}
/* ------------------------------------------------------------ */
/**
* Wraps an {@link ThreadPoolExecutor}.
* Max pool size is 256, pool thread timeout after 60 seconds, and core pool size is 32 when queueSize >= 0.
* @param queueSize can be -1 for using an unbounded {@link LinkedBlockingQueue}, 0 for using a
* {@link SynchronousQueue}, greater than 0 for using a {@link ArrayBlockingQueue} of the given size.
*/
public ExecutorThreadPool(int queueSize)
{
this(queueSize < 0 ? new ThreadPoolExecutor(256, 256, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()) :
queueSize == 0 ? new ThreadPoolExecutor(32, 256, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()) :
new ThreadPoolExecutor(32, 256, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueSize)));
}
/* ------------------------------------------------------------ */
/**
* Wraps an {@link ThreadPoolExecutor} using
* an unbounded {@link LinkedBlockingQueue} is used for the jobs queue;
* @param corePoolSize must be equal to maximumPoolSize
* @param maximumPoolSize the maximum number of threads to allow in the pool
* @param keepAliveTime the max time a thread can remain idle, in milliseconds
*/
public ExecutorThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime)
{
this(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MILLISECONDS);
}
/* ------------------------------------------------------------ */
/**
* Wraps an {@link ThreadPoolExecutor} using
* an unbounded {@link LinkedBlockingQueue} is used for the jobs queue.
* @param corePoolSize must be equal to maximumPoolSize
* @param maximumPoolSize the maximum number of threads to allow in the pool
* @param keepAliveTime the max time a thread can remain idle
* @param unit the unit for the keepAliveTime
*/
public ExecutorThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit)
{
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, new LinkedBlockingQueue<Runnable>());
}
/* ------------------------------------------------------------ */
/**
* Wraps an {@link ThreadPoolExecutor}
* @param corePoolSize the number of threads to keep in the pool, even if they are idle
* @param maximumPoolSize the maximum number of threads to allow in the pool
* @param keepAliveTime the max time a thread can remain idle
* @param unit the unit for the keepAliveTime
* @param workQueue the queue to use for holding tasks before they are executed
*/
public ExecutorThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)
{
this(new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue));
}
/* ------------------------------------------------------------ */
@Override
public void execute(Runnable job)
{
_executor.execute(job);
}
/* ------------------------------------------------------------ */
public boolean dispatch(Runnable job)
{
try
{
_executor.execute(job);
return true;
}
catch(RejectedExecutionException e)
{
LOG.warn(e);
return false;
}
}
/* ------------------------------------------------------------ */
public int getIdleThreads()
{
if (_executor instanceof ThreadPoolExecutor)
{
final ThreadPoolExecutor tpe = (ThreadPoolExecutor)_executor;
return tpe.getPoolSize() - tpe.getActiveCount();
}
return -1;
}
/* ------------------------------------------------------------ */
public int getThreads()
{
if (_executor instanceof ThreadPoolExecutor)
{
final ThreadPoolExecutor tpe = (ThreadPoolExecutor)_executor;
return tpe.getPoolSize();
}
return -1;
}
/* ------------------------------------------------------------ */
public boolean isLowOnThreads()
{
if (_executor instanceof ThreadPoolExecutor)
{
final ThreadPoolExecutor tpe = (ThreadPoolExecutor)_executor;
// getActiveCount() locks the thread pool, so execute it last
return tpe.getPoolSize() == tpe.getMaximumPoolSize() &&
tpe.getQueue().size() >= tpe.getPoolSize() - tpe.getActiveCount();
}
return false;
}
/* ------------------------------------------------------------ */
public void join() throws InterruptedException
{
_executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
}
/* ------------------------------------------------------------ */
@Override
protected void doStop() throws Exception
{
super.doStop();
_executor.shutdownNow();
}
}
|
