1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

private void enqueue(E x) {final Object[] items = this.items;items[putIndex] = x; // 入队if (++putIndex == items.length) // 如果指针到了末尾putIndex = 0; // 下一个入队的位置变为0count++;notEmpty.signal(); // 提醒消费者线程消费
}
private E dequeue() {final Object[] items = this.items;@SuppressWarnings("unchecked")E x = (E) items[takeIndex];items[takeIndex] = null; // 出队置空if (++takeIndex == items.length) // 如果指针到了末尾takeIndex = 0; // 下一个出队的位置变为0count--;if (itrs != null)itrs.elementDequeued();notFull.signal(); // 提醒生产者线程生产return x;
}

1
2
3
4
5

public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory);
}
public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory));
}

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

public class ThreadPoolTest {private static class Task implements Runnable {private int taskId;Task(int taskId) {this.taskId = taskId;}@Override public void run() {try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException ignore) {}System.out.println("task " + taskId + " end");}}public static void main(String[] args) {ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 2, 0, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(2) {@Override public boolean offer(Runnable runnable) {try {super.put(runnable); // 使用阻塞的put重写offer方法} catch (InterruptedException e) {e.printStackTrace();}return true;}});threadPool.submit(new Task(1));System.out.println("task 1 submitted");threadPool.submit(new Task(2));System.out.println("task 2 submitted");threadPool.submit(new Task(3));System.out.println("task 3 submitted");threadPool.submit(new Task(4));System.out.println("task 4 submitted");threadPool.submit(new Task(5));System.out.println("task 5 submitted");threadPool.submit(new Task(6));System.out.println("task 6 submitted");threadPool.shutdown();}}

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

public class ThreadPoolTest {private static class Task implements Runnable {private int taskId;Task(int taskId) {this.taskId = taskId;}@Overridepublic void run() {try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException ignore) {}System.out.println("task " + taskId + " end");}}private static class BlockCallerPolicy implements RejectedExecutionHandler {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {try {executor.getQueue().put(r);} catch (InterruptedException e) {e.printStackTrace();}}}public static void main(String[] args) {ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 2, 0, TimeUnit.SECONDS,new ArrayBlockingQueue<>(2), new BlockCallerPolicy());threadPool.submit(new Task(1));System.out.println("task 1 submitted");threadPool.submit(new Task(2));System.out.println("task 2 submitted");threadPool.submit(new Task(3));System.out.println("task 3 submitted");threadPool.submit(new Task(4));System.out.println("task 4 submitted");threadPool.submit(new Task(5));System.out.println("task 5 submitted");threadPool.submit(new Task(6));System.out.println("task 6 submitted");threadPool.shutdown();}}

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

public class StandardThreadExecutor extends LifecycleMBeanBaseimplements Executor, ResizableExecutor {// Tomcat线程池默认的配置protected int threadPriority = Thread.NORM_PRIORITY;protected boolean daemon = true;protected String namePrefix = "tomcat-exec-";protected int maxThreads = 200;protected int minSpareThreads = 25;protected int maxIdleTime = 60000;...protected boolean prestartminSpareThreads = false;protected int maxQueueSize = Integer.MAX_VALUE;protected void startInternal() throws LifecycleException {// 任务队列:这里你看到的是一个无界队列，但是队列里面进行了特殊处理taskqueue = new TaskQueue(maxQueueSize);TaskThreadFactory tf = new TaskThreadFactory(namePrefix,daemon,getThreadPriority());// 创建线程池，这里的ThreadPoolExecutor是Tomcat继承自JDK的ThreadPoolExecutorexecutor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), // 核心线程数与最大线程数maxIdleTime, TimeUnit.MILLISECONDS, // 默认6万毫秒的超时时间，也就是一分钟taskqueue, tf); // 玄机在任务队列的设置executor.setThreadRenewalDelay(threadRenewalDelay);if (prestartminSpareThreads) {executor.prestartAllCoreThreads(); // 预热所有的核心线程}taskqueue.setParent(executor);setState(LifecycleState.STARTING);}...
}

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

public class ThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor {private final AtomicInteger submittedCount = new AtomicInteger(0);private final AtomicLong lastContextStoppedTime = new AtomicLong(0L);private final AtomicLong lastTimeThreadKilledItself = new AtomicLong(0L);@Overrideprotected void afterExecute(Runnable r, Throwable t) {submittedCount.decrementAndGet(); // 执行完成后提交数量减一if (t == null) {// 如果有必要抛个异常让线程终止stopCurrentThreadIfNeeded();}}@Overridepublic void execute(Runnable command) {execute(command,0,TimeUnit.MILLISECONDS);}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)) {// 由于TaskQueue默认无界，所以默认强制入队会成功submittedCount.decrementAndGet();throw new RejectedExecutionException("Queue capacity is full.");}} catch (InterruptedException x) {submittedCount.decrementAndGet(); // 任务被拒绝，任务数减一throw new RejectedExecutionException(x);}} else {submittedCount.decrementAndGet(); // 任务被拒绝，任务数减一throw rx;}}}
}

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

public class TaskQueue extends LinkedBlockingQueue<Runnable> {private volatile ThreadPoolExecutor parent = null;public boolean force(Runnable o) {if ( parent==null || parent.isShutdown() )throw new RejectedExecutionException("Executor not running, can't force a command into the queue");// 因为LinkedBlockingQueue无界，所以调用offer强制入队return super.offer(o);}public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {if ( parent==null || parent.isShutdown() )throw new RejectedExecutionException("Executor not running, can't force a command into the queue");return super.offer(o,timeout,unit);}@Overridepublic boolean offer(Runnable o) {// 不是上面Tomcat中定义地ThreadPoolExecutor，不做任何检查if (parent==null) return super.offer(o);// 线程数达到最大线程数，尝试入队if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);// 提交的任务数小于线程数，也就是有空余线程，入队让空闲线程取任务if (parent.getSubmittedCount() < parent.getPoolSize()) return super.offer(o);// 走到这说明线程池没有空闲线程// 这里返回false，改变了LinkedBlockingQueue默认的行为// 使得Tomcat可以创建临时线程if (parent.getPoolSize() < parent.getMaximumPoolSize()) return false;// 到这里说明临时线程也没有空闲，只能排队了return super.offer(o);}
}