Java并发编程实战教程之线程池的饱和策略

18-06-28 来源：[db:作者]

收藏我要投稿

通常情况下，线程池会判断线程池的线程是否都处于工作状态。如果没有，则创建一个新的工作线程来执行任务。但是如果有限队列已经满了，则会交给饱和策略来处理这个任务。

ThreadPoolExecutor 的饱和策略可以通过调用 setRejectedExecutionHandler 来修改。JDK 提供了下面几种不同的饱和策略。

AbortPolicy CallerRunsPolicy DiscardPolicy DiscardOldestPolicy

下面我们举例来逐个来说明，先看下面代码，我们创建一个线程池，然后创建一个函数 getPolicy() 用于获取不同的饱和策略。通过传入不同的饱和策略参数变量选择不同的策略，通过执行结果来检查饱和策略的效果。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolRejectedPolicyTest{

 //线程池大小
 private static final int THREAD_SIZE = 2;
 //任务数量
 private static final int TASK_SIZE = 5;
 //饱和策略
 private static final int ABORT_POLICY = 0;//Abort 策略
 private static final int CALLER_RUNS_POLICY = 1;//CallerRuns 策略
 private static final int DISCARD_POLICY = 2;//Discard策略
 private static final int DISCARD_OLDEST_POLICY = 3;//DiscardOlds策略

 //策略选择变量，改这个值选择不同的饱和策略
 private static int selectedPolicy = ABORT_POLICY;

 @SuppressWarnings("unchecked")
 public static void main(String[] args) {
  //阻塞队列只能容纳一个任务
  BlockingQueue Q = new LinkedBlockingDeque<>(1);
  //创建一个固定线程的线程池
  ThreadPoolExecutor service = new ThreadPoolExecutor(THREAD_SIZE,THREAD_SIZE,0L, TimeUnit.MILLISECONDS,Q);
  //设置线程池的饱和策略
  service.setRejectedExecutionHandler(getPolicy(selectedPolicy));
  //提交任务
  for(int i = 0; i < TASK_SIZE; i++) {
service.submit(new Task(i));
  }
  service.shutdown();
 }

 //任务
 static class Task implements Runnable {
  //任务ID
  private int taskId;
  public Task(int id) {
taskId = id;
  }

  @Override
  public void run() {
try {
 Thread.sleep(50);
} catch (InterruptedException e) {
 System.err.println("线程被中断" + e.getMessage());  
}
//打印当前任务信息
System.out.println("Task " + taskId + " is running on " + Thread.currentThread().getName());
  }
 }

 //饱和策略
 static RejectedExecutionHandler getPolicy(int policy) {
  switch (policy) {
  case ABORT_POLICY:
return new ThreadPoolExecutor.AbortPolicy();
  case CALLER_RUNS_POLICY:
return new ThreadPoolExecutor.CallerRunsPolicy();
  case DISCARD_POLICY:
return new ThreadPoolExecutor.DiscardPolicy();
  case DISCARD_OLDEST_POLICY:
return new ThreadPoolExecutor.DiscardOldestPolicy();
  default:
break;
  }

  return new ThreadPoolExecutor.AbortPolicy();
 }
}