码迷，mamicode.com
首页 > 编程语言 > 详细

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

时间：2015-02-09 20:22:20      阅读：134      评论：0      收藏：0      [点我收藏+]

标签：

从Java5开始，Java提供了自己的线程池。每次只执行指定数量的线程，java.util.concurrent.ThreadPoolExecutor 就是这样的线程池。以下是我的学习过程。

首先是构造函数签名如下：

  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 拒绝策略（添加任务失败后如何处理该任务）.

1、线程池刚创建时，里面没有一个线程。任务队列是作为参数传进来的。不过，就算队列里面有任务，线程池也不会马上执行它们。
2、当调用 execute() 方法添加一个任务时，线程池会做如下判断：
    a. 如果正在运行的线程数量小于 corePoolSize，那么马上创建线程运行这个任务；
    b. 如果正在运行的线程数量大于或等于 corePoolSize，那么将这个任务放入队列。
    c. 如果这时候队列满了，而且正在运行的线程数量小于 maximumPoolSize，那么还是要创建线程运行这个任务；
    d. 如果队列满了，而且正在运行的线程数量大于或等于 maximumPoolSize，那么线程池会抛出异常，告诉调用者“我不能再接受任务了”。
3、当一个线程完成任务时，它会从队列中取下一个任务来执行。
4、当一个线程无事可做，超过一定的时间（keepAliveTime）时，线程池会判断，如果当前运行的线程数大于 corePoolSize，那么这个线程就被停掉。所以线程池的所有任务完成后，它最终会收缩到 corePoolSize 的大小。
       这个过程说明，并不是先加入任务就一定会先执行。假设队列大小为 4，corePoolSize为2，maximumPoolSize为6，那么当加入15个任务时，执行的顺序类似这样：首先执行任务 1、2，然后任务3~6被放入队列。这时候队列满了，任务7、8、9、10 会被马上执行，而任务 11~15 则会抛出异常。最终顺序是：1、2、7、8、9、10、3、4、5、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。

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

  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

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

如果将我们要运行的线程数10改为11，则由于最大线程数6+线程队列大小4=10<11，则根据线程池工作原则，最后一个线程将被拒绝策略拒绝，将示例二的main方法改为如下：

  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等待直到有线程可以被移除。

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

技术分享

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

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

标签：

原文地址：http://my.oschina.net/u/1398304/blog/376827
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！