java线程池之ThreadPoolExecutor(二):任务入队列和任务丢弃

ThreadPool

A simple C++11 Thread Pool implementation

项目地址：https://gitcode.com/gh_mirrors/th/ThreadPool

免费下载资源

御前两把刀刀

9025人浏览 · 2017-09-18 10:56:17

御前两把刀刀 · 2017-09-18 10:56:17 发布

一、关于任务入队列

在上一篇【java线程池之ThreadPoolExecutor(一)】中，

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(5));

定义线程池用的是ArrayBlockingQueue，该队列是有容量限制的。譬如这里，队列里面同时只能存在5个任务排队等候。

其官方解释如下：

ArrayBlockingQueue<E>：一个由数组支持的有界阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序。队列的头部是在队列中存在时间最长的元素。队列的尾部是在队列中存在时间最短的元素。新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素。
这是一个典型的“有界缓存区”，固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。一旦创建了这样的缓存区，就不能再增加其容量。试图向已满队列中放入元素会导致操作受阻塞；试图从空队列中提取元素将导致类似阻塞。

除了ArrayBlockingQueue队列，java还为我们提供了其它的队列：

LinkedBlockingQueue<E> ：一个基于已链接节点的、范围任意的 blocking queue。此队列按 FIFO（先进先出）排序元素。队列的头部是在队列中时间最长的元素。队列的尾部是在队列中时间最短的元素。新元素插入到队列的尾部，并且队列获取操作会获得位于队列头部的元素。链接队列的吞吐量通常要高于基于数组的队列，但是在大多数并发应用程序中，其可预知的性能要低。

PriorityBlockingQueue<E>：一个无界阻塞队列，它使用与类 PriorityQueue 相同的顺序规则，并且提供了阻塞获取操作。虽然此队列逻辑上是无界的，但是资源被耗尽时试图执行 add 操作也将失败（导致 OutOfMemoryError）。此类不允许使用 null 元素。依赖自然顺序的优先级队列也不允许插入不可比较的对象（这样做会导致抛出 ClassCastException）。

DelayQueue<E extends Delayed>：Delayed 元素的一个无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的 Delayed 元素。如果延迟都还没有期满，则队列没有头部，并且 poll 将返回 null。当一个元素的 getDelay(TimeUnit.NANOSECONDS) 方法返回一个小于等于 0 的值时，将发生到期。即使无法使用 take 或 poll 移除未到期的元素，也不会将这些元素作为正常元素对待。例如，size 方法同时返回到期和未到期元素的计数。此队列不允许使用 null 元素。

......

等等不一一列举，感兴趣的同学自己去查看api文档。

现在把上一篇【java线程池之ThreadPoolExecutor(一)】中的创建线程池的代码修改如下：

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());

运行结果如下：

正在执行task==2
正在执行task==4
正在执行task==1
正在执行task==3
正在执行task==5
=========第1次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：5
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：10
已经执行完成的任务数目：0
=========第1次检测end===========
=========第2次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：5
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：10
已经执行完成的任务数目：0
=========第2次检测end===========
=========第3次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：5
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：10
已经执行完成的任务数目：0
=========第3次检测end===========
线程:task==2执行完毕
正在执行task==6
=========第4次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：5
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：9
已经执行完成的任务数目：1
=========第4次检测end===========
=========第5次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：5
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：9
已经执行完成的任务数目：1
=========第5次检测end===========
线程:task==3执行完毕
正在执行task==7

........

线程:task==9执行完毕
正在执行task==13
线程:task==11执行完毕
正在执行task==14
线程:task==7执行完毕
正在执行task==15
=========第11次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：5
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：0
已经执行完成的任务数目：10
=========第11次检测end===========
线程:task==13执行完毕
=========第12次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：4
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：0
已经执行完成的任务数目：11
=========第12次检测end===========
线程:task==14执行完毕
=========第13次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：3
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：0
已经执行完成的任务数目：12
=========第13次检测end===========
线程:task==6执行完毕
线程:task==12执行完毕
=========第14次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：1
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：0
已经执行完成的任务数目：14
=========第14次检测end===========

......

=========第20次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：1
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：0
已经执行完成的任务数目：14
=========第20次检测end===========
线程:task==15执行完毕
=========第21次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：0
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：0
已经执行完成的任务数目：15
=========第21次检测end===========

你会发现活跃线程数始终不会超过5个，因为这时候队列的容量没有限制，不管多少任务来了，排队便是，线程池中始终由5个核心线程重复执行。

二. 线程池中的任务丢弃

对于任务丢弃，ThreadPoolExecutor以内部类的形式实现了4个策略。分别是：
CallerRunsPolicy。提交任务的线程自己负责执行这个任务。
AbortPolicy。使Executor抛出异常，通过异常做处理。
DiscardPolicy。丢弃提交的任务。
DiscardOldestPolicy。丢弃掉队列中最早加入的任务。

在调用构造方法时，参数中未指定RejectedExecutionHandler情况下，默认采用AbortPolicy。

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(5));
		 try {
			for(int i=1;i<=20;i++){//执行20个任务，但第16个任务会报异常，从而被拒绝。
			     MyTask myTask = new MyTask("task=="+i);
			     executor.execute(myTask);
			 }
		} catch (Exception e) {
			e.printStackTrace();
		}
		 //该线程用于对线程池的检测
		 new Thread(new CheckTask(executor)).start();
		 executor.shutdown();

运行结果如下所示：

java.util.concurrent.RejectedExecutionException: Task com.thread.pool.MyTask@aeea66 rejected from java.util.concurrent.ThreadPoolExecutor@19eda2c[Running, pool size = 10, active threads = 10, queued tasks = 5, completed tasks = 0]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(Unknown Source)
	at java.util.concurrent.ThreadPoolExecutor.reject(Unknown Source)
	at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source)
	at com.thread.pool.ThreadPoolTest2.main(ThreadPoolTest2.java:32)
正在执行task==2
正在执行task==4
正在执行task==11
正在执行task==13
正在执行task==15
正在执行task==1
正在执行task==3
正在执行task==5
正在执行task==12
正在执行task==14
线程:task==1执行完毕
正在执行task==6
线程:task==11执行完毕
正在执行task==7
=========第1次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：10
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：3
已经执行完成的任务数目：2
=========第1次检测end===========
=========第2次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：10
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：3
已经执行完成的任务数目：2
=========第2次检测end===========
线程:task==15执行完毕
正在执行task==8
线程:task==6执行完毕
正在执行task==9
=========第3次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：10
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：1
已经执行完成的任务数目：4
=========第3次检测end===========

从第16个任务开始就报异常被中止了。

当然我们在创建线程池的时候可以指明直接丢掉过多的任务，如下：

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(5),new ThreadPoolExecutor.DiscardPolicy());
		 try {
			for(int i=1;i<=20;i++){//执行20个任务，但第16个任务开始就会被丢弃。
			     MyTask myTask = new MyTask("task=="+i);
			     executor.execute(myTask);
			 }
		} catch (Exception e) {
			e.printStackTrace();
		}
		 //该线程用于对线程池的检测
		 new Thread(new CheckTask(executor)).start();
		 executor.shutdown();

请注意，线程池的构造方法现在多出了一个参数new ThreadPoolExecutor.DiscardPolicy()。运行结果如下：

正在执行task==2
正在执行task==4
正在执行task==11
正在执行task==13
正在执行task==15
正在执行task==1
正在执行task==3
正在执行task==5
正在执行task==12
正在执行task==14
线程:task==14执行完毕
正在执行task==6
=========第1次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：10
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：4
已经执行完成的任务数目：1
=========第1次检测end===========
线程:task==3执行完毕
正在执行task==7
线程:task==15执行完毕
正在执行task==8
=========第2次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：10
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：2
已经执行完成的任务数目：3
=========第2次检测end===========
=========第3次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：10
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：2
已经执行完成的任务数目：3
=========第3次检测end===========
.......

线程:task==7执行完毕
=========第12次检测start===========
线程池中核心线程数：5
线程池中活跃线程数目：0
线程池中允许的最大线程数目：10
队列中等待执行的任务数目：0
已经执行完成的任务数目：15
=========第12次检测end===========

注意：这里找不到第16个任务执行情况，也就是说从第16个任务开始就被丢弃了，也不会报错。
还有其它任务丢弃策略，请自行查看api文档研究！

GitHub 加速计划 / th / ThreadPool

7.74 K

2.22 K

下载

A simple C++11 Thread Pool implementation

最近提交(Master分支：2 个月前 )

9a42ec13 - 9 年前

fcc91415 - 9 年前

GitCode 开源社区

旨在为数千万中国开发者提供一个无缝且高效的云端环境，以支持学习、使用和贡献开源项目。

更多推荐

[转载]在Windows环境下安装GNU Radio

转自：在Windows环境下安装GNURadio_恐弱智_新浪博客GNU Radio是用Python开发的，大部分开源的工程能够在Linux环境下运行良好，而Windows下却运行的很勉强，而且安装配置都很复杂。GNU Radio算是个例外了，不光提供了Windows的二进制安装，还有比较详细的说明。我是Python小白，所以折腾了好久才弄好，特意记录下来，免得以后再装还折腾。GNU Radio的

GitCode 开源社区

centOS 8 使用dnf安装Docker

DNF是什么？CentOS 8使用YUM软件包管理器版本v4.0.4。现在，该版本使用DNF(已删除YUM)。DNF是软件包管理器。它会在Linux发行版上安装，执行更新并删除软件包。使用DNF安装Docker跳过具有损坏依赖性的程序包一个有效的解决方案是使您的CentOS 8系统使用以下--nobest命令安装最符合条件的版本：sudo dnf install docker...

GitCode 开源社区

定时同步数据库表(mysql+linux+crontab)

sync.sh里面的参数需要改变，ip/username/password/database/tablesync.sh#!/bin/sh# Please change the IP and password of the data source db.# Then change the table name.filename=/home/nington/db/$(date +%Y-%m