线程池池使用进阶篇（任务调度优化）_pool.map put(task)

作者：小小林熬夜学编程 | 2024-03-05 05:54:34

踩

pool.map put(task)

一、优化方案流程

二、方案的关键点

三、方案的优缺点

线程池原生任务调度执行策略：任务队列先进先出的模式，也就是先放到线程池中的任务先被执行。这种模式存在一个弊端就是无法对任务的优先级进行调度。

一、优化方案流程

针对这个问题，我这边自行实现了一个调度方案，调度方案包括：

1.线程池线程任务调度交给其中一个核心线程，核心线程根据线程任务优先级进行任务调度；

2.低优先级任务调度是不能占满线程池，确保高优先级任务来的时候可以及时响应处理。

方案的大致流程示意图：

因方案主要对线程任务调度的优化，所以流程图对线程任务调度到线程池之前的流程进行详细绘制并说明（对线程池内部线程调度流程及流程暂时忽略）。

这里提两点：

1.任务调度方案实现了任务拒绝策略，当线程池任务队列满，任务拒绝策略是将任务重新放回二级任务队列。

2.任务调度根据线程池任务队列是否满为其中一个调度调节，可以有效避免因为线程池任务满导致的任务被拒绝的情况。

二、方案的关键点

1.外部任务进来是，先放到二级任务队列，该队列根据任务优先级及任务名称进行区分，保证不同优先级的任务分别排在不同的任务队列中；


    /**
     * 普通线程池任务队列-二级队列
     */
    private HashMap<Integer, HashMap<String, LinkedBlockingQueue<Runnable>>> mTaskMapQueue =
            new HashMap<Integer, HashMap<String, LinkedBlockingQueue<Runnable>>>();

同优先级任务根据任务名称，进行队列存储，这个设计主要是为了扩展线程池根据线程任务名进行同等级任务进行调度。

2.调度线程作为核心线程，可以根据二级任务队列中任务优先级以及线程池任务队列情况进行任务调度；

调度线程是一个死循环线程，复杂循环调度任务，调度过程如下：

当线程池队列满，暂时不进行任务调度，直到线程池任务队列有排队空间；当线程池任务队列未满，遍历二级任务队列，取高优先级任务，如果有高优先级任务，取一个任务到线程池中，结束本轮循环调度；如无高优先级任务，遍历低优先级任务，如果此时线程池空闲线程小于一定数量N，暂时不调度低优先级任务，结束本地循环调度；如果线程池空闲线程大于一定数量N,取一个任务到线程池中，结束本轮循环调度.完成一轮循环调度立即进行下一轮循环调度。

三、代码实现

代码实现：


package com.yx.demo;
 
import java.lang.ref.WeakReference;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.WeakHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
 
/**
 * ThreadPoolManager
 *
 * @author yx
 * @date 2019/5/17 13:57
 */
public class ThreadPoolManager {
 
    private static final String THREAD_NAME_COMMON = "threadPool-common";
 
    /**
     * 普通线程池
     */
    private ThreadPoolExecutor mPoolExecutor;
 
    /**
     * 普通线程池任务队列-二级队列
     */
    private HashMap<Integer, HashMap<String, LinkedBlockingQueue<Runnable>>> mTaskMapQueue =
            new HashMap<Integer, HashMap<String, LinkedBlockingQueue<Runnable>>>();
    /**
     * 任务提交到线程池的任务集合
     */
    private Map<String, List<WeakReference<Future<?>>>> mTaskMap;
    /**
     * 单例
     */
    private static ThreadPoolManager instance = null;
 
    /**
     * 普通线程池核心线程池的数量，同时能够执行的线程数量
     */
    private int mCommonCorePoolSize = Runtime.getRuntime().availableProcessors() * 2 + 1;
 
    /**
     * 非核心线程数
     */
    private int mNonCorePoolSize = 10;
    /**
     * 最大线程池数量
     */
    private int mMaximumPoolSize = mCommonCorePoolSize + mNonCorePoolSize;
 
    /**
     * 低优先级任务不占用线程数
     * mMaximumPoolSize > mCommonCorePoolSize
     */
    private int mSpareThreadSize = 5;
 
    /**
     * 存活时间,根据各种线程任务执行超时时间评估 （网络重连任务 超时）
     */
    private long mKeepAliveTime = 30;
    /**
     * 线程池任务队列
     */
    private final int MAX_QUEUE_LENGTH = 200;
 
    private long mRejectCount = 0;
 
    private LinkedBlockingQueue<Runnable> mQueue =
            new LinkedBlockingQueue<Runnable>(MAX_QUEUE_LENGTH);
 
    /**
     * 创建一个新的实例 ThreadPoolManager
     */
    private ThreadPoolManager() {
        mPoolExecutor = new ThreadPoolExecutor(
                mCommonCorePoolSize,
                mMaximumPoolSize,
                mKeepAliveTime,
                TimeUnit.SECONDS,
                //缓冲队列，用于存放等待任务，Linked的先进先出
                mQueue,
                //创建线程的工厂
                new ThreadFactory() {
                    private final AtomicInteger mCount = new AtomicInteger(1);
 
                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, THREAD_NAME_COMMON + "#" + mCount.getAndIncrement());
                    }
                }
        );
        // 用来对超出maximumPoolSize的任务的处理策略
        mPoolExecutor.setRejectedExecutionHandler(new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
                mRejectCount++;
                System.out.println("ThreadPoolManager rejectedExecution happaned mRejectCount=" +
                        mRejectCount);
                ThreadPoolManager.getInstance().addCommonTask("", r, Thread.NORM_PRIORITY);
            }
        });
 
        mTaskMap = new WeakHashMap<String, List<WeakReference<Future<?>>>>();
        mPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        int available = mMaximumPoolSize - mPoolExecutor.getActiveCount();
                        // 线程池队列满了不调度
                        if (mQueue.size() >= MAX_QUEUE_LENGTH) {
                            continue;
                        }
                        // 调度策略 根据任务的优先级进行调度,高优先级任务优先调度
                        for (Integer p = Thread.MAX_PRIORITY; p > Thread.MIN_PRIORITY; p--) {
                            // 如果优先级小于5的线程(低优先级)来了，空闲线程太少，不调度
                            if (p < Thread.NORM_PRIORITY &&
                                    available < mSpareThreadSize) {
                                System.out
                                        .println("available = " + available + " mMaximumPoolSize=" +
                                                mMaximumPoolSize);
                                break;
                            }
                            HashMap<String, LinkedBlockingQueue<Runnable>> mapQueue =
                                    mTaskMapQueue.get(p);
                            if (mapQueue == null) {
                                continue;
                            }
                            if (!mapQueue.isEmpty()) {
                                Set<String> taskKeySet = mapQueue.keySet();
                                Iterator<String> subIterator = taskKeySet.iterator();
                                if (subIterator.hasNext()) {
                                    String taskName = subIterator.next();
                                    LinkedBlockingQueue<Runnable> queue = mapQueue.get(taskName);
                                    if (!queue.isEmpty()) {
                                        try {
                                            Runnable runnable = queue.take();
                                            ThreadPoolManager.getInstance()
                                                    .submitTask(runnable, taskName);
                                            // 正常调度了一个任务，重新遍历寻找高优先级任务
                                            break;
                                        } catch (InterruptedException e) {
                                            e.printStackTrace();
                                            break;
                                        }
                                    }
                                }
                            }
                            // 没有找到任务，继续找任务
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });
    }
 
    /**
     * 根据优先级加入二级队列
     *
     * @param taskName 任务名称
     * @param task     任务
     * @param priority 任务优先级
     * @throws NullPointerException
     */
    public synchronized void addCommonTask(String taskName, Runnable task, int priority) throws
            NullPointerException {
        if (priority > Thread.MAX_PRIORITY || priority < Thread.MIN_PRIORITY) {
            throw new IllegalArgumentException(
                    "priority must be  between " + Thread.MIN_PRIORITY + " and " +
                            Thread.MAX_PRIORITY);
        }
 
        if (taskName == null) {
            throw new NullPointerException("taskName can not be null");
        }
        HashMap<String, LinkedBlockingQueue<Runnable>> mapQueue = mTaskMapQueue.get(priority);
        if (mapQueue == null) {
            mapQueue = new HashMap<String, LinkedBlockingQueue<Runnable>>(16);
            LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
            mapQueue.put(taskName, queue);
        }
        LinkedBlockingQueue<Runnable> queue = mapQueue.get(taskName);
        queue.add(task);
        mTaskMapQueue.put(priority, mapQueue);
    }
 
    /**
     * 获取ThreadPoolManager单例对象
     */
    public static ThreadPoolManager getInstance() {
        if (instance == null) {
            synchronized (ThreadPoolManager.class) {
                if (instance == null) {
                    instance = new ThreadPoolManager();
                }
            }
        }
        return instance;
    }
 
    /**
     * 释放MinaThreadPoolManager中线程资源
     * 主要用于退出应用进程前，销毁线程池
     */
    public void release() {
        synchronized (ThreadPoolManager.class) {
            if (instance != null) {
                instance.cancelAllTaskThreads();
            }
            mPoolExecutor.shutdownNow();
            instance = null;
        }
    }
 
    /**
     * 将线程任务放到二级任务队列
     *
     * @param task 任务线程
     * @param name 任务名字
     */
    public void startTaskThread(Runnable task, String name) {
        if (task != null && name != null) {
            addCommonTask(name, task, Thread.NORM_PRIORITY);
        }
    }
 
    /**
     * 将线程任务放到二级任务队列
     *
     * @param task 任务线程
     * @param name 任务名字
     */
    public void startTaskThread(Runnable task, String name, int priority) {
        if (task != null && name != null) {
            addCommonTask(name, task, priority);
        }
    }
 
    /**
     * 开启线程
     *
     * @param task 任务线程
     * @param name 任务名字
     */
    public void submitTask(Runnable task, String name) {
        if (task != null && name != null) {
            Future<?> request = mPoolExecutor.submit(task);
            addTask(request, name);
            printPoolExecutorInfo();
        }
    }
 
    /**
     * 开启线程
     *
     * @param task 任务线程
     */
    @Deprecated
    public void executeTaskThread(Thread task) {
        String taskName = task.getName();
        System.out.println("ThreadPoolManager executeTaskThread task name = " + taskName);
        printPoolExecutorInfo();
        mPoolExecutor.execute(task);
    }
 
    /**
     * 结束线程
     *
     * @param task 任务线程
     */
    public void stopTaskThread(Thread task) {
        stopTaskThread(task.getName());
    }
 
    /**
     * 结束线程
     *
     * @param taskTag 任务线程
     */
    public void stopTaskThread(String taskTag) {
        cancelTaskThreads(taskTag);
    }
 
    /**
     * 添加执行任务到队列中
     *
     * @param request
     */
    private void addTask(Future<?> request, String taskTag) {
        synchronized (ThreadPoolManager.class) {
            if (taskTag != null) {
                List<WeakReference<Future<?>>> requestList = mTaskMap.get(taskTag);
                if (requestList == null) {
                    requestList = new LinkedList<WeakReference<Future<?>>>();
                    mTaskMap.put(taskTag, requestList);
                }
                requestList.add(new WeakReference<Future<?>>(request));
            }
        }
    }
 
    /**
     * 取消所有的任务
     */
    public void cancelAllTaskThreads() {
        for (String clsName : mTaskMap.keySet()) {
            List<WeakReference<Future<?>>> requestList = mTaskMap.get(clsName);
            if (requestList != null) {
                Iterator<WeakReference<Future<?>>> iterator = requestList.iterator();
                while (iterator.hasNext()) {
                    Future<?> request = iterator.next().get();
                    if (request != null) {
                        request.cancel(true);
                    }
                }
            }
        }
        mTaskMap.clear();
    }
 
    /**
     * 根据特定任务名称取消任务
     */
    private void cancelTaskThreads(String taskName) {
        System.out.println("ThreadPoolManager cancelTaskThreads task name = " + taskName);
        // 二级队列
        for (Integer p = Thread.MAX_PRIORITY; p > Thread.MIN_PRIORITY; p--) {
            HashMap<String, LinkedBlockingQueue<Runnable>> mapQueue =
                    mTaskMapQueue.get(p);
            if (mapQueue != null) {
                LinkedBlockingQueue<Runnable> queue = mapQueue.get(taskName);
                if (queue != null) {
                    queue.clear();
                }
            }
        }
        // 一级队列
        List<WeakReference<Future<?>>> requestList = mTaskMap.get(taskName);
        if (requestList != null) {
            Iterator<WeakReference<Future<?>>> iterator = requestList.iterator();
            while (iterator.hasNext()) {
                Future<?> request = iterator.next().get();
                if (request != null) {
                    request.cancel(true);
                }
            }
            mTaskMap.remove(taskName);
        }
        printPoolExecutorInfo();
    }
 
    public ThreadPoolExecutor getPoolExecutor() {
        return mPoolExecutor;
    }
 
    private void printPoolExecutorInfo() {
        if (mPoolExecutor != null) {
            System.out.println(
                    "ThreadPoolManager mPoolExecutor info:[poolSize:" + mPoolExecutor.getPoolSize()
                            + "，activeCount:" + mPoolExecutor.getActiveCount()
                            + "，taskQueueCount:" + mPoolExecutor.getQueue().size()
                            + "，completeTaskCount：" + mPoolExecutor.getCompletedTaskCount() + "]");
        }
    }
 
    /**
     * 任务队列中是否有
     *
     * @param task 任务
     * @return 是否有标签任务 true-有
     */
    public boolean hasTask(Thread task) {
        if (task == null) {
            return false;
        }
        return hasTask(task.getName());
    }
 
    /**
     * 任务队列中是否有
     *
     * @param taskTag 任务标签
     * @return 是否有标签任务 true-有
     */
    public boolean hasTask(String taskTag) {
        if (taskTag == null) {
            return false;
        }
        List<WeakReference<Future<?>>> requestList = mTaskMap.get(taskTag);
        if (requestList != null) {
            Iterator<WeakReference<Future<?>>> iterator = requestList.iterator();
            while (iterator.hasNext()) {
                Future<?> request = iterator.next().get();
                if (request != null) {
                    if (request.isCancelled()) {
                        System.out.println(
                                "ThreadPoolManager taskTag: " + taskTag + " has canceled ");
                        continue;
                    }
                    if (!request.isDone()) {
                        System.out.println("ThreadPoolManager hasTask " + taskTag);
                        return true;
                    }
                }
            }
        }
        return false;
    }
}

三、方案的优缺点

优点：.实现了任务调度根据优先级进行调度（暂未实现根据任务名调度）。

缺点：线程任务来到时，先放入二级任务队列，所以这个方案对于任务大量积压（比如任务大量添加，但线程池处理不及时）会对内存是一个挑战，所以使用时需要对低优先级耗时任务的添加进行限制，避免出现任务积压导致内存不足问题。

最后，关于方案还在持续优化中，如有问题，可以一起讨论交流。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/189731