ThreadPoolExecutor线程池以及线程复用原理

1
2
3
4
5
6
7
8
9
10
11
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
        BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);

1
2
3
4
5
    //阻塞队列
    BlockingQueue<Runnable> workQueue = null;
    workQueue = new ArrayBlockingQueue<>(5);//基于数组的先进先出队列，有界
    workQueue = new LinkedBlockingQueue<>();//基于链表的先进先出队列，无界
    workQueue = new SynchronousQueue<>();//无缓冲的等待队列，无界

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
 public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
  
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

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
//线程启动的时候
private boolean addWorker(Runnable firstTask, boolean core) {
  Worker w = null;
  w = new Worker(firstTask);
  final Thread t = w.thread;
  ...//代码
  t.start();
  ...//代码
}
        
------------------------------------------------------------------
 private final class Worker
            extends AbstractQueuedSynchronizer
            implements Runnable

        /** Thread this worker is running in.  Null if factory fails. */
        final Thread thread;
        /** Initial task to run.  Possibly null. */
        Runnable firstTask;
        /** Per-thread task counter */
        volatile long completedTasks;

        // worker内部类
        /**
         * Creates with given first task and thread from ThreadFactory.
         * @param firstTask the first task (null if none)
         */
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            // new一个工作线程
            this.thread = getThreadFactory().newThread(this);
        }
 
        
        // 重写run方法
        @override
        public void run() {
            runWorker(this);
        }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
//2，继续看里面的runWorker(this);很显然this是内部类Work对象本身
final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        ../代码
        try {
            while (task != null || (task = getTask()) != null) {
            try{
                ../代码
                task.run();
                }catch(){
                ../代码
                } finally {
                    ../代码
                }
            }
            ../代码
        } finally {
            // 从线程池回收线程
            processWorkerExit(w, completedAbruptly);
        }
 }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
private Runnable getTask() {
       ...//代码
       // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        
        try {
            // 从队列中获取runnable任务
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

1
在processWorkerExit方法中有进行worker线程移除 workers.remove(w);