相关概念
任务队列
- 如果线程池处理的任务小于当前提交任务，即当前任务处理不过来时，
将处理不了的任务放在队列中等待处理
拒绝策略
- 任务队列的数量不可能无限多，当处理的任务过于多，会受到处理能力的限制，导致性能低
- 常见的策略有：
抛出异常、直接丢弃、阻塞、放到临时队列
线程数目指标
init（min）
- 初始化线程时的创建的线程数目
active
- 运行时的一共的线程数，当长时间线程不够时，就会继续创建线程;当长时间不使用时，会销毁几个线程
max
- 是允许创建的最多线程数
<mark>max>=active>=min</mark>
版本一
需求：
- 实现一个简单的任务队列
- 实现初始化线程数目
任务队列实现
- 设计一个
队列，存放着Runnable实例，这些实例就是任务 - 任务队列的读写操作在并发的情况下存在线程不安全的问题，如超出最毒的队列数目，或者多个线程读取到一个任务，所以
必须加锁 - 每个线程都会尝试从任务队列中读取任务，如果抢到了任务队列中的任务，则执行该任务；如果没有抢到，则继续争抢，直到全部的任务执行完毕，陷入阻塞
- 给外界一个提交任务的接口，可以将Runnable任务写入到队列中（注意加锁），同时唤醒全部阻塞的线程
初始化线程数目实现
设计一个默认线程数，如果没有输入时，采用默认线程数在构造方法调用初始化方法，将根据数目创建线程，并放入到线程集合中，方便后续的管理
拒绝策略实现
- 设置一个属性保存最大的任务队列数，如果没有赋值，默认为2000
- 设计一个接口用于让用户自定义拒绝策略，如果没有设置自定义的，则
默认使用抛出异常 - 在提交任务的时候，判断是否超出了最大任务数。
如果超出，则执行拒绝策略
实现销毁线程池
- 判断任务队列是否为空，如果不为空，则休眠一会继续判断
- 设置一个变量用于保存还没有消除的线程数量
- 判断线程状态，
如果处于阻塞状态，则打断令其销毁 - 为了防止线程刚刚修改状态还没有进入等待状态，即
会出现打断失败，打断的同时需要将其状态设置为死亡状态
线程池的的扩容与空闲回收
- 线程池的扩容以及回收，都需要两个重要的参数：
当前的线程数、任务队列的大小 - 线程的扩容和回收应该
由线程池自动处理，线程池就是一个线程，当没有销毁时，每隔一段时间判断一次 - 需要特别注意锁的控制：如扩容时，可能会与销毁出现冲突，所以需要加锁；回收时与防止线程突然唤醒接到任务正在执行
public class SimpleThreadPool extends Thread{
private int size;//初始化线程数目
private final int queueSize;//最大任务队列大小
private final DiscardPolicy policy;//拒绝策略
private volatile boolean destroy = false;//是否是销毁状态
private int min;
private int max;
private int active;
private final Object Lock = new Object();
private static volatile int seq;//自增数字，模仿线程的默认命名方式
private static final String THREAD_PREFIX = "THREAD_POOL";//线程名词前缀
private final static ThreadGroup GROUP = new ThreadGroup("POOL_THREAD_GROUP");
private final static LinkedList<Runnable> TASK_QUEUE = new LinkedList<>();//任务队列，装着要执行的任务
private final static List<WorkerTask> THREAD_LIST = new ArrayList<>();//线程组
private final static int DEFAULT_TASK_QUEUE_SIZE = 2000;//线程队列默认最大值
private final static DiscardPolicy DEFAULT_DISCARD_POLICY = ()->{//默认拒绝策略
throw new DiscardException("拒绝该任务！！！");
};
public SimpleThreadPool() {
this( 4,8,12,DEFAULT_TASK_QUEUE_SIZE,DEFAULT_DISCARD_POLICY);
}
public SimpleThreadPool(int queuesize){
this(4,8,12,queuesize,DEFAULT_DISCARD_POLICY);
}
public SimpleThreadPool(int min,int active,int max,int queueSize,DiscardPolicy policy) {//初始化线程数大小
this.queueSize = queueSize;
this.policy = policy;
this.min = min;
this.active = active;
this.max = max;
this.init();
}
/** * 初始化线程池 */
private void init() {
for (int i = 0; i < min; i++) {
this.CreatWorkerTask();
}
this.size = min;
this.start();
}
/** * 创建线程 */
private void CreatWorkerTask() {
WorkerTask workerTask = new WorkerTask(GROUP, THREAD_PREFIX + (seq++));
workerTask.start();
THREAD_LIST.add(workerTask);
}
/** * 进行扩容，以及输出当前线程池的状态 */
@Override
public void run() {
while (!destroy){
System.out.printf("Pool#Min:%d,Active:%d,Max:%d,Current:%d,QueueSize:%d\n",
this.min, this.active, this.max, this.size, TASK_QUEUE.size());
try {
Thread.sleep(5_000);
synchronized (Lock){
if (TASK_QUEUE.size()>active && size <active && !destroy){//线程扩容到active
for (int i = size; i < active; i++) {
this.CreatWorkerTask();
}
System.out.println("线程池扩容到----active");
size = active;
}else if (TASK_QUEUE.size()>max && size <max && !destroy){//线程扩容到max
for (int i = size; i < max; i++) {
this.CreatWorkerTask();
}
System.out.println("线程池扩容到----max");
size = max;
}
if (TASK_QUEUE.isEmpty() && size > active && !destroy){//线程降级到active
synchronized (TASK_QUEUE){
int release = size - active;
for (Iterator<WorkerTask> it = THREAD_LIST.iterator();it.hasNext();) {
if (release<=0){
break;
}
WorkerTask wt = it.next();
wt.close();
wt.interrupt();
it.remove();
release--;
}
size = active;
}
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
/** * 提交任务功能，同时唤醒正在阻塞的队列 * * @param runnable 提交的任务 */
public void submit(Runnable runnable) {
if (destroy){
throw new IllegalStateException("线程池已经被销毁，不能再提交任务");
}
synchronized (TASK_QUEUE) {
if (TASK_QUEUE.size()>queueSize){//当超出规定的最大任务数
policy.discard();//执行拒绝策略
}else {
TASK_QUEUE.addLast(runnable);
TASK_QUEUE.notifyAll();//唤醒全部阻塞线程
}
}
}
/** * 销毁线程池 * @throws InterruptedException */
public void shutdown() throws InterruptedException {
synchronized (Lock){
while (!TASK_QUEUE.isEmpty()){
Thread.sleep(50);
}
int initVal = THREAD_LIST.size();
while (initVal>0){
for (WorkerTask task : THREAD_LIST) {
if (task.getTaskState()==TaskState.BLOCKED){
task.interrupt();
task.close();//防止线程还没有wait的时候被打断，无法跳出循环
initVal--;
}else {
Thread.sleep(10);
}
}
}
this.destroy =true;
System.out.println("线程池已经关闭");
}
}
public int getSize() {
return size;
}
public int getQueuesize() {
return queueSize;
}
public boolean isDestroy(){
return this.destroy;
}
private enum TaskState {//线程生命周期的一些状态
FARE, RUNNING, BLOCKED, DEDE
}
/** * 拒绝策略抛出的异常 */
public static class DiscardException extends RuntimeException {
public DiscardException(String message) {
super(message);
}
}
/** * 拒绝策略 */
public interface DiscardPolicy {
void discard() throws DiscardException;
}
/** * 自定义的工作线程 * 用于执行工作队列中的任务 */
private static class WorkerTask extends Thread {
private volatile TaskState taskState = TaskState.FARE;//默认是空闲状态
public WorkerTask(ThreadGroup group, String name) {
super(group, name);
}
public TaskState getTaskState() {
return this.taskState;
}
@Override
public void run() {
OUTER:
while (this.taskState != TaskState.DEDE) {//只要线程没有死亡
Runnable runnable;
synchronized (TASK_QUEUE) {
while (TASK_QUEUE.isEmpty()) {
try {
taskState = TaskState.BLOCKED;//此时没有任务，线程陷入阻塞
TASK_QUEUE.wait();
} catch (InterruptedException e) {
taskState = TaskState.DEDE;
break OUTER;//直接跳出多级循环，防止唤醒后，线程已经死亡
}
}
runnable = TASK_QUEUE.removeFirst();
}
if (runnable != null) {//如果可以运行
taskState = TaskState.RUNNING;
runnable.run();
taskState = TaskState.FARE;
}
}
}
public void close() {//关闭线程
this.taskState = TaskState.DEDE;
}
}
/** * 测试 */
public static void main(String[] args) throws InterruptedException {
SimpleThreadPool pool = new SimpleThreadPool();
Thread.sleep(5_000);
new Thread(()->{
for (int i = 0; i < 200; i++) {
int finalI = i;
pool.submit(() -> {
System.out.println("当前是" + Thread.currentThread().getName() + "线程为我服务......" );
try {
Thread.sleep(1_000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("当前是" + Thread.currentThread().getName() + "线程完成了服务！！！" + finalI);
});
}
}).start();
Thread.sleep(10_000);
pool.shutdown();
}
}
京公网安备 11010502036488号