`

关于ThreadPoolExecutor 调用RejectedExecutionHandler的机制

    博客分类：
  • Java
java线程 
阅读更多

当我们创建线程池并且提交任务失败时，线程池会回调RejectedExecutionHandler接口的rejectedExecution(Runnable task, ThreadPoolExecutor executor)方法来处理线程池处理失败的任务，其中task 是用户提交的任务，而executor是当前执行的任务的线程池。可以通过代码的方式来验证。

1、线程池工厂：

 

?
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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
package com.threadpool;
 
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
 
/**
 * 线程池工厂方法
 * @author
 *
 */
public class ThreadPoolFactory {
     
    //线程池
    private static ThreadPoolExecutor  pool;   
    //自身对象
    private static ThreadPoolFactory factory;
     
    /**
     * 私有构造函数
     */
    private ThreadPoolFactory(){    }
     
    /**
     * 获取工厂对象
     * @param config
     * @return
     */
    public static ThreadPoolFactory getInstance(ThreadPoolConfig config){
        if(factory == null){
            factory = new ThreadPoolFactory();
        }
         
        if(pool == null){
             
            if(config.getHandler() == null){
                pool = new ThreadPoolExecutor(config.getCorePoolSize(),
                        config.getMaximumPoolSize(),config.getKeepAliveTime(),
                        config.getUnit(),config.getWorkQueue());
            }else{
                pool = new ThreadPoolExecutor(config.getCorePoolSize(),
                        config.getMaximumPoolSize(),config.getKeepAliveTime(),
                        config.getUnit(),config.getWorkQueue(),config.getHandler());
            }
        }      
        System.out.println("pool  create= "+pool.toString());
        return factory;
    }
     
    /**
     * 添加线程池任务
     * @param run
     */
    public synchronized void addTask(Runnable run){
        pool.execute(run);
    }
     
    /**
     * 添加线程池任务
     * @param runs
     */
    public synchronized void addTask(List<Runnable> runs){
        if(runs != null){
            for(Runnable r:runs){
                this.addTask(r);
            }
        }
    }
     
    /**
     * 关闭线程池
     */
    public void closePool(){
        pool.shutdown();
    }
     
}

2、线程池配置文件类：

 

 

?
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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
package com.threadpool;
 
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.TimeUnit;
 
/**
 * 线程池配置类
 * @author
 *
 */
public class ThreadPoolConfig {
    //池中所保存的线程数，包括空闲线程。
    private int corePoolSize;
    //池中允许的最大线程数。
    private int maximumPoolSize;
    //当线程数大于核心时，此为终止前多余的空闲线程等待新任务的最长时间。
    private long keepAliveTime;
    //参数的时间单位。
    private TimeUnit unit;
    //执行前用于保持任务的队列。此队列仅由保持 execute 方法提交的 Runnable 任务。
    private BlockingQueue<Runnable> workQueue;
    //由于超出线程范围和队列容量而使执行被阻塞时所使用的处理程序。
    private RejectedExecutionHandler handler;
    //配置文件自身对象
    private static ThreadPoolConfig config;
    /**
     * 单例模式
     */
    private ThreadPoolConfig(){
         
    }
     
    /**
     * 获取配置文件对象
     * @return
     */
    public static ThreadPoolConfig getInstance(){
        if(config == null){
            config = new ThreadPoolConfig();
        }      
        return config;
    }  
    public int getCorePoolSize() {
        return corePoolSize;
    }
    public void setCorePoolSize(int corePoolSize) {
        this.corePoolSize = corePoolSize;
    }
    public int getMaximumPoolSize() {
        return maximumPoolSize;
    }
    public void setMaximumPoolSize(int maximumPoolSize) {
        this.maximumPoolSize = maximumPoolSize;
    }
    public long getKeepAliveTime() {
        return keepAliveTime;
    }
    public void setKeepAliveTime(long keepAliveTime) {
        this.keepAliveTime = keepAliveTime;
    }
    public TimeUnit getUnit() {
        return unit;
    }
    public void setUnit(TimeUnit unit) {
        this.unit = unit;
    }
    public BlockingQueue<Runnable> getWorkQueue() {
        return workQueue;
    }
    public void setWorkQueue(BlockingQueue<Runnable> workQueue) {
        this.workQueue = workQueue;
    }
    public RejectedExecutionHandler getHandler() {
        return handler;
    }
    public void setHandler(RejectedExecutionHandler handler) {
        this.handler = handler;
    }  
}

3、简单任务类：

 

 

?
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
package com.test;
 
/**
 * 任务线程
 * @author
 *
 */
public class ThreadTask extends Thread {
     
    public ThreadTask(String name){
        super(name);
    }
     
    @SuppressWarnings("static-access")
    @Override
    public void run() {
        // TODO Auto-generated method stub
        System.out.println(this.getName().toString() + ", will sleep 0 s");
        try {
            this.sleep(1*10);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.out.println(this.getName().toString() + ", I am wakeup now ");
    }
 
}

4、异常处理接口实现类：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
package com.threadpool;
 
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
 
/**
 * 线程池异常处理类
 * @author
 *
 */
public class MyRejectedExecutionHandler implements RejectedExecutionHandler {
 
    @Override
    public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
        // TODO Auto-generated method stub
        System.out.println("Begin exception handler-----------");
        //执行失败任务
        new Thread(task,"exception by pool").start();
        //打印线程池的对象
        System.out.println("The pool RejectedExecutionHandler = "+executor.toString());
    }
}

5、测试主函数：

?
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
package com.test;
 
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
 
import com.threadpool.MyRejectedExecutionHandler;
import com.threadpool.ThreadPoolConfig;
import com.threadpool.ThreadPoolFactory;
 
/**
 * @author
 *
 */
public class TestThreadPoolMain {
 
    /**
     * @param args
     */
    public static void main(String[] args) {
 
        //设置配置
        ThreadPoolConfig config = ThreadPoolConfig.getInstance();
        config.setCorePoolSize(2);
        config.setMaximumPoolSize(3);
        config.setKeepAliveTime(5);
        config.setUnit(TimeUnit.SECONDS);
        //将队列设小，会抛异常
        config.setWorkQueue(new ArrayBlockingQueue<Runnable>(10));
        config.setHandler(new MyRejectedExecutionHandler());
        //线程池工厂
        ThreadPoolFactory factory = ThreadPoolFactory.getInstance(config);
         
        for(int i = 0;i<100;i++){
            factory.addTask(new ThreadTask(i+"-i"));
        }
        System.out.println("i add is over!-------------------");
    }
}

6、测试比较：

可以看出创建的线程池对象和调用传递的线程池对象是相同的。

pool create = java.util.concurrent.ThreadPoolExecutor@de6f34
0-i, will sleep 0 s
Begin exception handler-----------
12-i, will sleep 0 s
The pool RejectedExecutionHandler = java.util.concurrent.ThreadPoolExecutor@de6f34
Begin exception handler-----------
1-i, will sleep 0 s
The pool RejectedExecutionHandler = java.util.concurrent.ThreadPoolExecutor@de6f34

分享到：
| 视频的播放
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    使用线程池ThreadPoolExecutor 抓取论坛帖子列表

    4. 当线程数达到最大值且任务队列已满，新提交的任务会被拒绝，此时可以设置一个`RejectedExecutionHandler`来处理这种情况，例如丢弃任务或抛出异常。 在实际应用中，我们可能还需要关注线程池的监控和调优，例如...

    java中ThreadPoolExecutor常识汇总

    * RejectedExecutionHandler：饱和策略，最大线程和工作队列容量且已经饱和时 execute 方法都将调用 RejectedExecutionHandler ThreadPoolExecutor 的工作流程 ThreadPoolExecutor 的工作流程可以分为以下四步： ...

    Java线程池的拒绝策略实现详解

    最后，让我们通过一个简单的示例代码来演示如何使用ThreadPoolExecutor和RejectedExecutionHandler接口来实现线程池的拒绝策略。 ```java package com.cfang; import java.util.concurrent.BlockingQueue; import ...

    Android线程池管理的代码例子

    RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy(); // 拒绝策略 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( corePoolSize, maximumPoolSize, keepAliveTime,...

    java并发编程学习笔记

    `ThreadPoolExecutor`通过`BlockingQueue`的阻塞机制来维持线程池的稳定运行。具体来说，当线程池中的线程没有任务可执行时，会调用`workQueue.take()`方法阻塞等待新任务。 - **具体步骤**： 1. 每次提交任务时，...

    java线程池

    线程池的构造方法`ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue&lt;Runnable&gt; workQueue, RejectedExecutionHandler handler)`中的参数有着至关重要...

    Java线程池学习资料-全

    使用线程池时，我们需要考虑线程池的参数，如核心线程数（corePoolSize）、最大线程数（maxPoolSize）、工作队列（WorkQueue）、拒绝策略（RejectedExecutionHandler）和空闲线程存活时间（keepAliveTime）。...

    线程池之Executor框架.docx

    Executor框架的设计理念是将任务（工作单元）与执行机制分离，从而提高了程序的可扩展性和灵活性。 1. **Executor框架的核心组件** - **任务**：Executor框架处理的主要对象是任务，任务可以是实现`Runnable`接口...

    Java并发编程：线程池的使用 - 平凡希 - 博客园1

    `ThreadPoolExecutor`类提供了四个构造方法，它们的主要区别在于是否指定了`ThreadFactory`和`RejectedExecutionHandler`。所有构造方法都需要以下几个关键参数： 1. `corePoolSize`：核心线程池大小，这是线程池的...

    Java多线程按指定顺序同步执行

    join方法可以使得某个线程A调用join，其他线程就要乖乖等A执行完毕才能执行。示例代码如下： ```java public class Worker implements Runnable { private int number; public Worker(int i) { number = i; } ...

    定制化线程池实现高并发数据处理.zip

    在Java编程中，线程池是一种管理线程的机制，它可以有效地提高系统资源的利用率，减少线程创建和销毁的开销，同时还能控制系统的并发程度。本资料"定制化线程池实现高并发数据处理"主要探讨如何通过自定义线程池来...

    线程池Demo(待续)

    然后在`Test.java`文件中，可能会调用线程池来执行这些自定义线程： ```java executor.execute(new CustomThread()); ``` 线程池的优势在于它能有效地避免频繁地创建和销毁线程，这在处理大量并发请求时尤为关键。...

    java线程池的使用方式

    RejectedExecutionHandler handler) ``` - **corePoolSize**：线程池的基本大小，在任何时间都会维持这么多线程。 - **maximumPoolSize**：线程池允许的最大线程数量。 - **keepAliveTime**：当线程数超过基本大小...

    socket 线程池实现

    RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy(); ExecutorService executor = new ThreadPoolExecutor( corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, ...

    day07【线程池、Lambda表达式】(1).md

    - `RejectedExecutionHandler`：拒绝策略，当线程池无法接受新任务时的处理方式。 #### 三、Lambda 表达式 ##### 3.1 Lambda 表达式概念 - **定义**：Lambda 表达式是一种匿名函数，它可以像普通对象一样被传递...

    Java concurrency线程池之线程池原理(二)_动力节点Java学院整理

    如果已经达到最大线程数，且任务队列已满（无界队列情况下不会发生），新提交的任务将会被拒绝，具体行为取决于`RejectedExecutionHandler`。 在示例中，我们创建了5个线程对象并提交到线程池，由于线程池大小为2，...

    java 线程池

    Java线程池是一种高效管理并发任务执行的机制，它通过维护一组可重用的线程来减少创建和销毁线程的开销。线程池在Java中由`java.util.concurrent`包下的`ExecutorService`接口及其实现类，尤其是`ThreadPoolExecutor...

    ThradPoolDemo

    7. **拒绝策略（RejectedExecutionHandler）**：当线程池和队列都满时，处理新任务的策略，常见的有抛出异常、直接忽略、调用系统钩子或停止整个线程池。 在`ThradPoolDemo`中，可能会展示如何通过`...

    线程池使用示例（含源代码）

    RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy(); ExecutorService executor = new ThreadPoolExecutor( corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.SECONDS, ...

Magicbox
Global site tag (gtag.js) - Google Analytics