开发者社区> 韩曙亮> 正文

【Android 异步操作】线程池 ( 线程池 reject 拒绝任务 | 线程池 addWorker 添加任务 )

简介: 【Android 异步操作】线程池 ( 线程池 reject 拒绝任务 | 线程池 addWorker 添加任务 )
+关注继续查看

文章目录

一、线程池 reject 拒绝任务

二、线程池 addWorker 添加任务



在上一篇博客 【Android 异步操作】线程池 ( 线程池 execute 方法源码解析 ) 中 , 讲解 线程池 ThreadPoolExecutor 的 execute 方法时 , 有两个重要的核心方法 ;


两个核心的操作 :


添加任务 : addWorker(command, true) , 第二个参数为 true 是添加核心线程任务 , 第二个参数为 false 是添加非核心线程任务 ;

拒绝任务 : reject(command)

本博客中研究 拒绝任务 reject 方法的细节 ;






一、线程池 reject 拒绝任务


在 ThreadPoolExecutor 线程池中 , void reject(Runnable command) 方法 , 主要是调用了 RejectedExecutionHandler handler 的 rejectedExecution 方法 ;


该 handler 可以在如下 构造函数中传入 , 如下构造函数中的最后一个参数 ;


 

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {


部分相关代码示例 :


public class ThreadPoolExecutor extends AbstractExecutorService {
    /**
     * 当线程池任务饱和 , 或线程池关闭 , 使用该 Handler 处理拒绝任务异常情况 ;
     */
    private volatile RejectedExecutionHandler handler;
    /**
     * 在该构造函数中 , 可以传入 Handler ;
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
   
    /**
     * 拒绝执行给定命令的处理 
     */
    final void reject(Runnable command) {
        handler.rejectedExecution(command, this);
    }
}



自定义 RejectedExecutionHandler 处理拒绝任务的情况 ;


实现 rejectedExecution 方法 , 当 线程池任务队列饱和 , 或者 没有空闲线程 时 , 线程池被关闭 时 , 导致线程池的任务队列无法接受该任务时 , 会调用该方法



/**
 * 处理添加任务失败情况的 handler 
 * 用户可以自定义该接口 
 */
public interface RejectedExecutionHandler {
    /**
     * 
     * 无法接受任务时执行该方法 ;
     * 当线程池任务队列饱和 , 或者没有空闲线程时 , 线程池被关闭时 , 会调用该方法
     *
     * 该方法可能会触发 RejectedExecutionException 异常 , 用户需要捕获并处理该异常
     *
     * @param r 被拒绝的任务
     * @param executor 尝试执行该任务的执行者
     * @throws RejectedExecutionException 如果没有补救方法, 抛出该异常
     */
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}






二、线程池 addWorker 添加任务


检查一个新的工作者 ( Worker ) 是否可以被添加 , 根据当前的 线程池状态 , 和 给定的核心线程数 , 最大线程数 等判定 ;


如果可以添加 , 那么先 调整 工作者 ( Worker ) 的个数 , 然后 创建新的 工作者 ( Worker ) , 并 将参数中的 Runnable firstTask 设置为第一个任务 ;


如果线程池停止或关闭 , 返回 false ;


如果线程创建失败 , 不管是线程工厂返回空 , 还是出现 OOM , 直接退出 ;



如果当前的 线程个数少于 核心线程数 , 或者当前的 任务数已满 , 则必须创建 工作者 ( Worker ) , 并执行第一个初始任务 ,


public class ThreadPoolExecutor extends AbstractExecutorService {
    /**
     * 检查一个新的工作者是否可以被添加 , 根据当前的线程池状态 , 和给定的核心线程数 , 最大线程数等判定 ;
     * 如果可以添加 , 那么先调整 工作者 ( Worker ) 的个数 , 然后创建新的 工作者 ( Worker ) , 
     * 并将参数中的 Runnable firstTask 设置为第一个任务 ; 
     * 如果线程池停止或关闭 , 返回 false ; 
     * 如果线程创建失败 , 不管是线程工厂返回空 , 还是出现 OOM , 直接退出 ; 
     *
     * 第一个运行的任务 ; 
     * 如果当前的线程个数少于 核心线程数 , 或者当前的任务数已满 , 
     * 必须创建 工作者 ( Worker ) , 并执行第一个初始任务 , 
     *
     * @param core 如果设置为 true , 必须使用核心线程绑定 
     * @return true if successful
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        // 死循环
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            // 查看队列是否为空
            // 查看线程池是否 SHUTDOWN 
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
    // 查看线程池的线程个数 , 与 核心线程数, 最大线程数 进行各种对比
    // 获取现在的线程池情况 
            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
    // 如果没有达到最大线程数 , 允许添加 
                if (compareAndIncrementWorkerCount(c))  // 该行代码只是将计数器 +1
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
  // 下面是添加一个线程的逻辑  
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            // 创建工作者 , 并将任务设置给该工作者 
            w = new Worker(firstTask);
            // 线程是从工作者中取出的 , 该线程是在 Worker 构造函数中使用线程工厂创建的 
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
      // 将工作者设置给 工作者集合 
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                // 如果添加成功 , 就会启动 Worker 中的线程 
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
}





版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
Java多线程--线程常用操作方法
<h1>1、取得和设置线程名称</h1> <div> <img src="http://img.blog.csdn.net/20131011094925062?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvd2FsZG1lcg==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/
1334 0
Java多线程--线程操作范例
<h1>1、实例要求</h1> <div> <span style="white-space:pre"></span>设计一个线程操作类,要求可以产生三个线程对象,并可以分别设置三个线程的休眠时间:</div> <div> <span style="white-space:pre"></span>线程A : 休眠10秒</div> <div> <span style="white-s
898 0
8.9 Linux系统添加新用户(useradd命令)
Linux 系统中,可以使用 useradd 命令新建用户,此命令的基本格式如下:
26 0
SpringBoot项目使用多线程处理任务时无法通过Autowired注入bean
  最近在做一个“温湿度控制”的项目,项目要求通过用户设定的温湿度数值和实时采集到的数值进行比对分析,因为数据的对比与分析是一个通过前端页面控制的定时任务,经理要求在用户开启定时任务时,单独开启一个线程进行数据的对比分析,并将采集到的温湿度数值存入数据库中的历史数据表,按照我们正常的逻辑应该是用户在请求开启定时任务时,前端页面通过调用后端接口,创建一个新的线程来执行定时任务,然后在线程类中使用 @Autowired 注解注入保存历史数据的service层,在线程类中调用service层保存历史数据的方法实现温湿度数据的保存,这时就出现了一个很尴尬的问题,在新开启的线程中使用 @Autowire
59 0
操作系统的线程管理机制以及Thread.Sleep()的作用
说明:本文章内容来自某论坛里的一位大牛的回复。 引题:我们可能经常会用到 Thread.Sleep 函数来使线程挂起一段时间。
740 0
Linux基础命令---添加用户useradd
useradd       创建新的系统用户,useradd指令只能以管理员的身份运行,创建的用户都在“/etc/passwd”文件中。当不加-D参数,useradd指令使用命令列来指定新帐号的设定值and使用系统上的预设值.新使用者帐号将产生一些系统档案,使用者目录建立,拷备起始档案等,这些均可以利用命令列选项指定。
1335 0
Linux下使用两个线程协作完成一个任务的简易实现
刚解决了之前的那个Linux下Pthread库的问题,这次就来使用两个线程来协作,共同完成一个求和的任务。 打一下基础吧 本次需要使用到的知识点有: lpthread,编译的时候用到,相当于一个声明的效果 pthread_mutex_t 作为一个锁对象而存在 pth...
952 0
+关注
韩曙亮
专注 Android 领域
2605
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
JS零基础入门教程(上册)
立即下载
性能优化方法论
立即下载
手把手学习日志服务SLS,云启实验室实战指南
立即下载