JAVA线程池ThreadPoolExecutor与阻塞队列BlockingQueue .

时间  2020-12-03
标签 java ide 函数 学习 测试 this spa .net 线程 code 栏目 Java 繁體版
原文   原文链接
从Java5开始，Java提供了本身的线程池。每次只执行指定数量的线程，java.util.concurrent.ThreadPoolExecutor 就是这样的线程池。如下是个人学习过程。

首先是构造函数签名以下： java

  1. public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);    
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
参数介绍：

corePoolSize 核心线程数，指保留的线程池大小（不超过maximumPoolSize值时，线程池中最多有corePoolSize 个线程工做）。 
maximumPoolSize 指的是线程池的最大大小（线程池中最大有corePoolSize 个线程可运行）。 
keepAliveTime 指的是空闲线程结束的超时时间（当一个线程不工做时，过keepAliveTime 长时间将中止该线程）。 
unit 是一个枚举，表示 keepAliveTime 的单位（有NANOSECONDS, MICROSECONDS, MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS，7个可选值）。 
workQueue 表示存听任务的队列（存放须要被线程池执行的线程队列）。 
handler 拒绝策略（添加任务失败后如何处理该任务）. ide

一、线程池刚建立时，里面没有一个线程。任务队列是做为参数传进来的。不过，就算队列里面有任务，线程池也不会立刻执行它们。
二、当调用 execute() 方法添加一个任务时，线程池会作以下判断：
    a. 若是正在运行的线程数量小于 corePoolSize，那么立刻建立线程运行这个任务；
    b. 若是正在运行的线程数量大于或等于 corePoolSize，那么将这个任务放入队列。
    c. 若是这时候队列满了，并且正在运行的线程数量小于 maximumPoolSize，那么仍是要建立线程运行这个任务；
    d. 若是队列满了，并且正在运行的线程数量大于或等于 maximumPoolSize，那么线程池会抛出异常，告诉调用者“我不能再接受任务了”。
三、当一个线程完成任务时，它会从队列中取下一个任务来执行。
四、当一个线程无事可作，超过必定的时间（keepAliveTime）时，线程池会判断，若是当前运行的线程数大于 corePoolSize，那么这个线程就被停掉。因此线程池的全部任务完成后，它最终会收缩到 corePoolSize 的大小。
       这个过程说明，并非先加入任务就必定会先执行。假设队列大小为 4，corePoolSize为2，maximumPoolSize为6，那么当加入15个任务时，执行的顺序相似这样：首先执行任务 一、2，而后任务3~6被放入队列。这时候队列满了，任务七、八、九、10 会被立刻执行，而任务 11~15 则会抛出异常。最终顺序是：一、二、七、八、九、十、三、四、五、6。固然这个过程是针对指定大小的ArrayBlockingQueue<Runnable>来讲，若是是LinkedBlockingQueue<Runnable>，由于该队列无大小限制，因此不存在上述问题。 函数

示例一，LinkedBlockingQueue<Runnable>队列使用： 学习

  1. import java.util.concurrent.BlockingQueue;  
  2. import java.util.concurrent.LinkedBlockingQueue;  
  3. import java.util.concurrent.ThreadPoolExecutor;  
  4. import java.util.concurrent.TimeUnit;  
  5.   
  6. /** 
  7.  * Created on 2011-12-28 
  8.  * <p>Description: [Java 线程池学习]</p> 
  9.  * @author         shixing_11@sina.com 
  10.  */  
  11. public class ThreadPoolTest implements Runnable {   
  12.      public void run() {   
  13.           synchronized(this) {   
  14.             try{  
  15.                 System.out.println(Thread.currentThread().getName());  
  16.                 Thread.sleep(3000);  
  17.             }catch (InterruptedException e){  
  18.                 e.printStackTrace();  
  19.             }  
  20.           }   
  21.      }   
  22.        
  23.      public static void main(String[] args) {   
  24.          BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();  
  25.          ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);  
  26.          for (int i = 0; i < 10; i++) {     
  27.              executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));     
  28.              int threadSize = queue.size();  
  29.              System.out.println("线程队列大小为-->"+threadSize);  
  30.          }     
  31.          executor.shutdown();    
  32.      }  
  33. }    
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created on 2011-12-28
 * <p>Description: [Java 线程池学习]</p>
 * @author         shixing_11@sina.com
 */
public class ThreadPoolTest implements Runnable { 
     public void run() { 
          synchronized(this) { 
            try{
                System.out.println(Thread.currentThread().getName());
                Thread.sleep(3000);
            }catch (InterruptedException e){
                e.printStackTrace();
            }
          } 
     } 
     
     public static void main(String[] args) { 
         BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
         for (int i = 0; i < 10; i++) {   
             executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));   
             int threadSize = queue.size();
             System.out.println("线程队列大小为-->"+threadSize);
         }   
         executor.shutdown();  
     }
}
输出结果以下：

线程队列大小为-->0
线程名称：pool-1-thread-1
线程队列大小为-->0
线程队列大小为-->1
线程队列大小为-->2
线程队列大小为-->3
线程队列大小为-->4
线程队列大小为-->5
线程队列大小为-->6
线程队列大小为-->7
线程队列大小为-->8
线程名称：pool-1-thread-2
线程名称：pool-1-thread-1
线程名称：pool-1-thread-2
线程名称：pool-1-thread-1
线程名称：pool-1-thread-2
线程名称：pool-1-thread-1
线程名称：pool-1-thread-2
线程名称：pool-1-thread-1
线程名称：pool-1-thread-2 测试

可见，线程队列最大为8，共执行了10个线线程。由于是从线程池里运行的线程，因此虽然将线程的名称设为"TestThread".concat(""+i)，但输出后仍是变成了pool-1-thread-x。 this

示例二，LinkedBlockingQueue<Runnable>队列使用： spa

  1. import java.util.concurrent.BlockingQueue;  
  2. import java.util.concurrent.ArrayBlockingQueue;  
  3. import java.util.concurrent.ThreadPoolExecutor;  
  4. import java.util.concurrent.TimeUnit;  
  5.   
  6. /** 
  7.  * Created on 2011-12-28 
  8.  * <p>Description: [Java 线程池学习]</p> 
  9.  * @author         shixing_11@sina.com 
  10.  */  
  11. public class ThreadPoolTest implements Runnable {   
  12.      public void run() {   
  13.           synchronized(this) {   
  14.             try{  
  15.                 System.out.println("线程名称："+Thread.currentThread().getName());  
  16.                 Thread.sleep(3000); //休眠是为了让该线程不至于执行完毕后从线程池里释放   
  17.             }catch (InterruptedException e){  
  18.                 e.printStackTrace();  
  19.             }  
  20.           }   
  21.      }   
  22.        
  23.      public static void main(String[] args) throws InterruptedException {   
  24.          BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列   
  25.          ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);  
  26.          for (int i = 0; i < 10; i++) {     
  27.              executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));     
  28.              int threadSize = queue.size();  
  29.              System.out.println("线程队列大小为-->"+threadSize);  
  30.          }     
  31.          executor.shutdown();    
  32.      }  
  33. }   
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created on 2011-12-28
 * <p>Description: [Java 线程池学习]</p>
 * @author         shixing_11@sina.com
 */
public class ThreadPoolTest implements Runnable { 
     public void run() { 
          synchronized(this) { 
            try{
                System.out.println("线程名称："+Thread.currentThread().getName());
                Thread.sleep(3000); //休眠是为了让该线程不至于执行完毕后从线程池里释放
            }catch (InterruptedException e){
                e.printStackTrace();
            }
          } 
     } 
     
     public static void main(String[] args) throws InterruptedException { 
         BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列
         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
         for (int i = 0; i < 10; i++) {   
             executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));   
             int threadSize = queue.size();
             System.out.println("线程队列大小为-->"+threadSize);
         }   
         executor.shutdown();  
     }
}
输出结果以下：

线程队列大小为-->0
线程队列大小为-->0
线程队列大小为-->1
线程队列大小为-->2
线程队列大小为-->3
线程队列大小为-->4
线程队列大小为-->4
线程队列大小为-->4
线程队列大小为-->4
线程名称：pool-1-thread-1
线程名称：pool-1-thread-3
线程名称：pool-1-thread-2
线程名称：pool-1-thread-4
线程队列大小为-->4
线程名称：pool-1-thread-5
线程名称：pool-1-thread-6
线程名称：pool-1-thread-5
线程名称：pool-1-thread-6
线程名称：pool-1-thread-4
线程名称：pool-1-thread-2 .net

可见，总共10个线程，由于核心线程数为2，2个线程被当即运行，线程队列大小为4，因此4个线程被加入队列，最大线程数为6，还能运行6-2=4个，其10个线程的其他4个线程又当即运行了。 线程

若是将咱们要运行的线程数10改成11，则因为最大线程数6+线程队列大小4=10<11，则根据线程池工做原则，最后一个线程将被拒绝策略拒绝，将示例二的main方法改成以下： code

  1. public static void main(String[] args) throws InterruptedException {   
  2.     ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列   
  3.     ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);  
  4.     for (int i = 0; i < 11; i++) {     
  5.         executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));     
  6.         int threadSize = queue.size();  
  7.         System.out.println("线程队列大小为-->"+threadSize);  
  8.     }     
  9.     executor.shutdown();    
  10. }   
public static void main(String[] args) throws InterruptedException { 
         ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列
         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
         for (int i = 0; i < 11; i++) {   
             executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));   
             int threadSize = queue.size();
             System.out.println("线程队列大小为-->"+threadSize);
         }   
         executor.shutdown();  
     }
输出结果：

线程队列大小为-->0
线程名称：pool-1-thread-1
线程队列大小为-->0
线程队列大小为-->1
线程队列大小为-->2
线程队列大小为-->3
线程队列大小为-->4
线程名称：pool-1-thread-2
线程队列大小为-->4
线程名称：pool-1-thread-3
线程队列大小为-->4
线程名称：pool-1-thread-4
线程队列大小为-->4
线程名称：pool-1-thread-5
线程队列大小为-->4
线程名称：pool-1-thread-6
Exception in thread "main" java.util.concurrent.RejectedExecutionException
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 ths.ThreadPoolTest.main(ThreadPoolTest.java:30)
线程名称：pool-1-thread-1
线程名称：pool-1-thread-3
线程名称：pool-1-thread-2
线程名称：pool-1-thread-4

很明显，抛RejectedExecutionException异常了，被拒绝策略拒绝了，这就说明线程超出了线程池的总容量（线程队列大小+最大线程数）。

         对于 java.util.concurrent.BlockingQueue 类有有三种方法将线程添加到线程队列里面，然而如何区别三种方法的不一样呢，其实在队列未满的状况下结果相同，都是将线程添加到线程队列里面，区分就在于当线程队列已经满的时候，此时

public boolean add(E e) 方法将抛出IllegalStateException异常，说明队列已满。

public boolean offer(E e) 方法则不会抛异常，只会返回boolean值，告诉你添加成功与否，队列已满，固然返回false。

public void put(E e) throws InterruptedException 方法则一直阻塞（即等待，直到线程池中有线程运行完毕，能够加入队列为止）。

为了证实对上面这三个方法的描述，咱们将示例二改成以下、public boolean add(E e)方法测试程序：

  1. public static void main(String[] args) throws InterruptedException {   
  2.     BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列   
  3.     ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);  
  4.     for (int i = 0; i < 10; i++) {     
  5.         executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));     
  6.         int threadSize = queue.size();  
  7.         System.out.println("线程队列大小为-->"+threadSize);  
  8.         if (threadSize==4){  
  9.             queue.add(new Runnable() {  //队列已满，抛异常   
  10.                 @Override  
  11.                 public void run(){  
  12.                     System.out.println("我是新线程，看看能不能搭个车加进去！");  
  13.                       
  14.                 }  
  15.             });  
  16.         }  
  17.     }     
  18.     executor.shutdown();    
  19. }   
public static void main(String[] args) throws InterruptedException { 
         BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列
         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
         for (int i = 0; i < 10; i++) {   
             executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));   
             int threadSize = queue.size();
             System.out.println("线程队列大小为-->"+threadSize);
             if (threadSize==4){
                 queue.add(new Runnable() {  //队列已满，抛异常
                     @Override
                     public void run(){
                         System.out.println("我是新线程，看看能不能搭个车加进去！");
                         
                     }
                 });
             }
         }   
         executor.shutdown();  
     }
运行结果：

线程队列大小为-->0
线程名称：pool-1-thread-1
线程队列大小为-->0
线程队列大小为-->1
线程队列大小为-->2
线程队列大小为-->3
线程队列大小为-->4
线程名称：pool-1-thread-2
Exception in thread "main" java.lang.IllegalStateException: Queue full
at java.util.AbstractQueue.add(Unknown Source)
at java.util.concurrent.ArrayBlockingQueue.add(Unknown Source)
at ths.ThreadPoolTest.main(ThreadPoolTest.java:35)
线程名称：pool-1-thread-1
线程名称：pool-1-thread-2
线程名称：pool-1-thread-2
线程名称：pool-1-thread-1

很明显，当线程队列已满，即线程队列里的线程数为4时，抛了异常，add线程失败。再来看public boolean offer(E e) 方法测试程序：

  1. <STRONG> </STRONG>public static void main(String[] args) throws InterruptedException {   
  2.          BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列   
  3.          ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);  
  4.          for (int i = 0; i < 10; i++) {     
  5.              executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));     
  6.              int threadSize = queue.size();  
  7.              System.out.println("线程队列大小为-->"+threadSize);  
  8.              if (threadSize==4){  
  9.                  final boolean flag = queue.offer(new Runnable() {  
  10.                      @Override  
  11.                      public void run(){  
  12.                          System.out.println("我是新线程，看看能不能搭个车加进去！");  
  13.                            
  14.                      }  
  15.                  });  
  16.                  System.out.println("添加新线程标志为-->"+flag);  
  17.              }  
  18.          }     
  19.          executor.shutdown();    
  20.      }   
 public static void main(String[] args) throws InterruptedException { 
         BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列
         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
         for (int i = 0; i < 10; i++) {   
             executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));   
             int threadSize = queue.size();
             System.out.println("线程队列大小为-->"+threadSize);
             if (threadSize==4){
                 final boolean flag = queue.offer(new Runnable() {
                     @Override
                     public void run(){
                         System.out.println("我是新线程，看看能不能搭个车加进去！");
                         
                     }
                 });
                 System.out.println("添加新线程标志为-->"+flag);
             }
         }   
         executor.shutdown();  
     }
运行结果以下：

线程队列大小为-->0
线程名称：pool-1-thread-1
线程队列大小为-->0
线程队列大小为-->1
线程队列大小为-->2
线程队列大小为-->3
线程队列大小为-->4
添加新线程标志为-->false
线程队列大小为-->4
线程名称：pool-1-thread-3
添加新线程标志为-->false
线程名称：pool-1-thread-2
线程队列大小为-->4
添加新线程标志为-->false
线程名称：pool-1-thread-4
线程队列大小为-->4
添加新线程标志为-->false
线程名称：pool-1-thread-5
线程队列大小为-->4
添加新线程标志为-->false
线程名称：pool-1-thread-6
线程名称：pool-1-thread-1
线程名称：pool-1-thread-2
线程名称：pool-1-thread-3
线程名称：pool-1-thread-4

能够看到，当线程队列已满的时候，线程没有被添加到线程队列，程序也没有抛异常。继续看public void put(E e) throws InterruptedException;方法测试程序：

  1. public static void main(String[] args) throws InterruptedException {   
  2.          BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列   
  3.          ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);  
  4.          for (int i = 0; i < 10; i++) {     
  5.              executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));     
  6.              int threadSize = queue.size();  
  7.              System.out.println("线程队列大小为-->"+threadSize);  
  8.              if (threadSize==4){  
  9.                  queue.put(new Runnable() {  
  10.                      @Override  
  11.                      public void run(){  
  12.                          System.out.println("我是新线程，看看能不能搭个车加进去！");  
  13.                      }  
  14.                  });  
  15.              }  
  16.          }     
  17.          executor.shutdown();    
  18.      }   
public static void main(String[] args) throws InterruptedException { 
         BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列
         ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
         for (int i = 0; i < 10; i++) {   
             executor.execute(new Thread(new ThreadPoolTest(), "TestThread".concat(""+i)));   
             int threadSize = queue.size();
             System.out.println("线程队列大小为-->"+threadSize);
             if (threadSize==4){
                 queue.put(new Runnable() {
                     @Override
                     public void run(){
                         System.out.println("我是新线程，看看能不能搭个车加进去！");
                     }
                 });
             }
         }   
         executor.shutdown();  
     }
结果以下：

线程队列大小为-->0
线程队列大小为-->0
线程队列大小为-->1
线程队列大小为-->2
线程队列大小为-->3
线程队列大小为-->4
线程名称：pool-1-thread-1
线程名称：pool-1-thread-2
线程名称：pool-1-thread-1
线程名称：pool-1-thread-2
线程队列大小为-->3
线程队列大小为-->4
线程名称：pool-1-thread-3
线程名称：pool-1-thread-2
线程队列大小为-->4
线程名称：pool-1-thread-1
线程队列大小为-->4
我是新线程，看看能不能搭个车加进去！
线程名称：pool-1-thread-3
线程名称：pool-1-thread-4
我是新线程，看看能不能搭个车加进去！
线程名称：pool-1-thread-3
我是新线程，看看能不能搭个车加进去！
我是新线程，看看能不能搭个车加进去！

很明显，尝试了四次才加进去，前面三次尝试添加，但因为线程sleep(3000)，因此没有执行完，线程队列一直处于满的状态，直到某个线程执行完，队列有空位，新线程才加进去，没空位以前一直阻塞（即等待），我能加进去为止。


那么线程池的排除策略是什么样呢，通常按以下规律执行：

A.  若是运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。
B.  若是运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。
C.  若是没法将请求加入队列，则建立新的线程，除非建立此线程超出 maximumPoolSize，在这种状况下，任务将被拒绝。


总结：

1. 线程池可当即运行的最大线程数 即maximumPoolSize 参数。

2. 线程池能包含的最大线程数 = 可当即运行的最大线程数 + 线程队列大小 (一部分当即运行，一部分装队列里等待)

3. 核心线程数可理解为建议值，即建议使用的线程数，或者依据CPU核数

4. add，offer，put三种添加线程到队列的方法只在队列满的时候有区别，add为抛异常，offer返回boolean值，put直到添加成功为止。

5.同理remove，poll， take三种移除队列中线程的方法只在队列为空的时候有区别， remove为抛异常，poll为返回boolean值， take等待直到有线程能够被移除。

看看下面这张图就清楚了：


记在这里作为学习的过程，之后偶尔有空翻起来容易。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程