欢迎访问昆山宝鼎软件有限公司网站! 设为首页 | 网站地图 | XML | RSS订阅 | 宝鼎邮箱 | 后台管理


新闻资讯

MENU

软件开发知识

队列中正在等待执行的任务数 CAD加密 量为:4线程池中现在的线程数目是:10

点击: 次  来源:宝鼎软件 时间:2017-12-19

原文出处: Sam哥哥

概述

在JAVA的世界里,假如想并行的执行一些任务,可以利用ThreadPoolExecutor。
大部门环境下直接利用ThreadPoolExecutor就可以满意要求了,可是在某些场景下,好比瞬时大流量的,为了提高响应和吞吐量,最好照旧扩展一下ThreadPoolExecutor。

全宇宙的JAVA IT人士应该都知道ThreadPoolExecutor的执行流程:

  • core线程还能应付的,则不绝的建设新的线程;
  • core线程无法应付,则将任务扔到行列内里;
  • 行列满了(意味着插入任务失败),则开始建设MAX线程,线程数到达MAX后,行列还一直是满的,则抛出RejectedExecutionException.
  • 这个执行流程有个小问题,就是当core线程无法应付请求的时候,会立即将任务添加到行列中,假如行列很是长,而任务又很是多,那么将会有频繁的任务入行列和任务出行列的操纵。

    按照实际的压测发明,这种操纵也是有必然耗损的。其实JAVA提供的SynchronousQueue行列是一个零长度的行列,任务都是直接由出产者递交给消费者,中间没有入行列的进程,可见JAVA API的设计者也是有思量过入行列这种操纵的开销。

    别的,昆山软件公司,昆山软件开发,任务一多,立即扔到行列里,而MAX线程又不干活,假如行列内里太多任务了,只有可怜的core线程在忙,也是会影响机能的。

    当core线程无法应付请求的时候,能不能延后入行列这个操纵呢? 让MAX线程尽快启动起来,资助处理惩罚任务。

    也等于说,当core线程无法应付请求的时候,假如当前线程池中的线程数量还小于MAX线程数的时候,继承建设新的线程处理惩罚任务,一直到线程数量达到MAX后,才将任务插入到行列里。

    我们通过包围行列的offer要领来实现这个方针。

      @Override
    public  boolean offer(Runnable o) {
        int currentPoolThreadSize = executor.getPoolSize();
        //假如线程池里的线程数量已经达到最大,将任务添加到行列中
        if (currentPoolThreadSize == executor.getMaximumPoolSize()) {
            return super.offer(o);
        }
        //说明有空闲的线程,这个时候无需建设core线程之外的线程,而是把任务直接丢到行列里即可
        if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {
            return super.offer(o);
        }
    
        //假如线程池里的线程数量还没有达到最大,直接建设线程,而不是把任务丢到行列内里
        if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
            return false;
        }
    
        return super.offer(o);
    }

    留意个中的

    if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {
            return super.offer(o);
    }

    是暗示core线程仍然能处理惩罚的来,同时又有空闲线程的环境,将任务插入到行列中。 如何判定线程池中有空闲线程呢? 可以利用一个计数器来实现,每当execute要领被执行的时候,计较器加1,当afterExecute被执行后,计数器减1.

        @Override
        public void execute(Runnable command) {
            submittedTaskCount.incrementAndGet();
            //代码未完整,待增补。。。。。
        }
     @Override
        protected void afterExecute(Runnable r, Throwable t) {
            submittedTaskCount.decrementAndGet();
        }

    这样,当

    executor.getSubmittedTaskCount() < currentPoolThreadSize

    的时候,说明有空闲线程。

    完整代码

    EnhancedThreadPoolExecutor类

    package executer;
    
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class EnhancedThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor {
    
        /**
         * 计数器,用于暗示已经提交到行列内里的task的数量,这里task特指还未完成的task。
         * 当task执行完后,submittedTaskCount会减1的。
         */
        private final AtomicInteger submittedTaskCount = new AtomicInteger(0);
    
        public EnhancedThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, TaskQueue workQueue) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new ThreadPoolExecutor.AbortPolicy());
            workQueue.setExecutor(this);
        }
    
        /**
         * 包围父类的afterExecute要领,当task执行完成后,将计数器减1
         */
        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            submittedTaskCount.decrementAndGet();
        }
    
    
        public int getSubmittedTaskCount() {
            return submittedTaskCount.get();
        }
    
    
        /**
         * 包围父类的execute要领,在任务开始执行之前,计数器加1。
         */
        @Override
        public void execute(Runnable command) {
            submittedTaskCount.incrementAndGet();
            try {
                super.execute(command);
            } catch (RejectedExecutionException rx) {
                //当产生RejectedExecutionException,实验再次将task丢到行列内里,假如照旧产生RejectedExecutionException,则直接抛出异常。
                BlockingQueue<Runnable> taskQueue = super.getQueue();
                if (taskQueue instanceof TaskQueue) {
                    final TaskQueue queue = (TaskQueue)taskQueue;
                    if (!queue.forceTaskIntoQueue(command)) {
                        submittedTaskCount.decrementAndGet();
                        throw new RejectedExecutionException("行列已满");
                    }
                } else {
                    submittedTaskCount.decrementAndGet();
                    throw rx;
                }
            }
        }
    }