ScheduleThreadPoolExecutor线程池分析

╰+哭是因爲堅強的太久メ 2022-05-11 03:48 144阅读 0赞

ScheduleThreadPoolExecutor是官方推荐的取代Timer作定时任务的线程池，在研究ScheduleThreadPoolExecutor过程中发现此线程池无论什么时候都**只会有核心线程数coreSize个线程在工作**。

这样就有个问题，如果任务较为繁重的情况下，或者单个任务执行时间较长的情况下，是否需要调整coreSize的大小。  
**先从ScheduleThreadPoolExecutor的构造函数出发：**  
只说最复杂构造函数了，其参数为corePoolSize核心线程数，threadFactory线程工厂，handler拒绝处理策略。构造函数中调用了父类ThreadPoolExecutor的构造函数,传入的参数中第二个参数为线程池最大线程数Integer.MAX\_VALUE，第三个参数线程保持时间为0，第五个参数为任务队列DelayedWorkQueue，这几个参数为默认传入参数。

public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory,
                                       RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue(), threadFactory, handler);
    }

从以上构造函数看，ScheduleThreadPoolExecutor的核心线程数可由构造函数传入，最大线程数为Integer的最大值，任务队列为DelayedWorkQueue。

研究DelayedWorkQueue类源码可知它就是一个小根堆，排序规则按的是任务执行时间，堆顶为最快执行的任务，内部数据结构是一个数组，由于Java是静态语言，数组满时扩容每次扩容50%。**通过siftUp和siftDown方法维持小根堆特征，时间复杂度为O（ln）**, **通过grow()进行扩容。**

/**
         * Sifts element added at bottom up to its heap-ordered spot.
         * Call only when holding lock.
         */
        private void siftUp(int k, RunnableScheduledFuture<?> key) {
            while (k > 0) {
                int parent = (k - 1) >>> 1;
                RunnableScheduledFuture<?> e = queue[parent];
                if (key.compareTo(e) >= 0)
                    break;
                queue[k] = e;
                setIndex(e, k);
                k = parent;
            }
            queue[k] = key;
            setIndex(key, k);
        }
    
        /**
         * Sifts element added at top down to its heap-ordered spot.
         * Call only when holding lock.
         */
        private void siftDown(int k, RunnableScheduledFuture<?> key) {
            int half = size >>> 1;
            while (k < half) {
                int child = (k << 1) + 1;
                RunnableScheduledFuture<?> c = queue[child];
                int right = child + 1;
                if (right < size && c.compareTo(queue[right]) > 0)
                    c = queue[child = right];
                if (key.compareTo(c) <= 0)
                    break;
                queue[k] = c;
                setIndex(c, k);
                k = child;
            }
            queue[k] = key;
            setIndex(key, k);
        }
        
    	/**
         * Resizes the heap array.  Call only when holding lock.
         */
        private void grow() {
            int oldCapacity = queue.length;
            int newCapacity = oldCapacity + (oldCapacity >> 1); // grow 50%
            if (newCapacity < 0) // overflow
                newCapacity = Integer.MAX_VALUE;
            queue = Arrays.copyOf(queue, newCapacity);
        }

从这里可以看出这是个无界队列，根据线程池线程增长流程，如果线程数小于核心线程数，会创建线程执行任务，如果线程数等于核心线程数，并且队列未满的情况下，放到队列中，并不会创建新线程，这样ScheduleThreadPoolExecutor将永远只会有核心线程数个线程。

下面为测试代码：

public class TestScheduleThreadPoolExecutor {
    
    public static final ScheduledThreadPoolExecutor stpe = new ScheduledThreadPoolExecutor(2);
    public static int index = 0;
    
    public static void main(String[] args) {
    
       Thread scheduleThread = new Thread(new Runnable() {
    
            @Override
            public void run() {
    
                while(true) {
                  
                    stpe.scheduleAtFixedRate(new Runnable() {
                        
                        @Override
                        public void run() {
        
                            System.out.println(index++);
                            try {
                                Thread.sleep(1000);
                            }
                            catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }  
                    }, 1, 10, TimeUnit.SECONDS);
                    
                    try {
                        Thread.sleep(200);
                    }
                    catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }  
       });
       
       scheduleThread.start();
       
       Thread recordThread = new Thread(new Runnable() {
    
           @Override
           public void run() {
    
               while(true) {
                 
                   int number = stpe.getActiveCount();
                   System.out.println("Thread number: "+ number + ", TaskCount: " + stpe.getQueue().size() +", LargestPoolSize: " + stpe.getLargestPoolSize());
                   try {
                       Thread.sleep(200);
                   }
                   catch (InterruptedException e) {
                       e.printStackTrace();
                   }
               }
           }  
      });  
       recordThread.start();
    }
    }

测试结果：

255
    Thread number: 2, QueueCount: 254, LargestPoolSize: 2
    256
    Thread number: 2, QueueCount: 255, LargestPoolSize: 2
    257
    Thread number: 2, QueueCount: 256, LargestPoolSize: 2
    258
    Thread number: 2, QueueCount: 257, LargestPoolSize: 2
    259
    Thread number: 2, QueueCount: 258, LargestPoolSize: 2
    260
    Thread number: 2, QueueCount: 259, LargestPoolSize: 2

在任务个数不断增长的情况下，有效线程数一直保持两个。这也说明线程一直有任务在运行。

**这说明如果定时任务太多，或者定时任务处理时间过长会造成任务累积。**

**这里应该就应该考虑到自动调整核心线程数的大小，具体这篇文章就不说了。**