JDK并发包（线程池）（1）

r囧r小猫 2022-05-17 02:29 237阅读 0赞

# 1.什么是线程池 #

为了避免系统频繁地创建和销毁线程，我们可以让创建的线程进行复用。如同数据库连接池一样当系统使用数据库时不是创建一个新的连接，而是从连接池中获取一个可用的连接，反之当需要关闭连接是，并不是真的进行关闭连接，而是将这个连接返还给连接池。通过这种方式，可以节约很多创建和销毁对象的时间。

线程池也是类似的概念。线程池中总有几个活跃的线程，当你需要使用线程时，可以从池中拿一个空闲线程，当完成工作时，并不关闭它而是将这个线程退回到线程池中方便重复利用。

![70][]

# 2.JDK对线程池的支持 #

为了能更好的控制多线程，JDK提供了一套Executor框架。

![70 1][]

以上成员均在java.util.concurrent包中，是JDK并发包的核心类。其中ThreadPoolExecutor表示一个线程池。Executors类则是线程池工厂的角色，通过Executors可以获取一个特定功能的线程池。从上面的UML图中可以看出，ThreadPoolExecutor类实现了Executor接口因此通过这个接口任何Runnable对象都可以被ThreadPoolExecutor线程池调度。

Executor框架提供了各种类型的线程池，主要有以下工厂方法：

1.  newFixedThreadPool()方法：这个方法会返回一个固定线程数量的线程池。该线程池中线程数量不变，当有任务时，如果这个线程池中由空余线程则立即执行，如果没有则会暂存到一个任务队列中，等有线程空闲时再执行。

2.newCachedThreadPool()方法：这个方法返回一个可根据实际情况调整线程数量的线程池。线程池的线程数不确定，但是如果有空闲线程可以使用则优先使用空闲线程。如果所有线程都在工作又有新任务提交时则新建线程处理任务。所有线程在当前任务执行完毕后将返回线程池进行复用。

3.newSingleThreadExecutor()方法：这个方法返回一个只有一个线程的线程池。如果多于一个任务被提交到该线程池，任务会被保存在一个任务队列中，等线程空闲时按先入先出的顺序执行任务队列中的任务。

4.newSingleThreadScheduledExecutor()方法：这个方法返回一个ScheduledExecutorService对象，线程池大小为1。ScheduledExecutorService接口在ExecutorService接口之上扩展了在给定时间执行某任务的功能，如在某个固定的延时后执行，或周期性执行某个任务。

5.newScheduledThreadPool()方法：该方法也返回一个ScheduledExecutorService对象，但是该线程池可以指定线程数量。

固定大小的线程池：

package thread.pool;
    
    import java.util.Date;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class ThreadPoolDemo {
    	public static class MyTask implements Runnable{
    
    		@Override
    		public void run() {
    			System.out.println(new Date(System.currentTimeMillis())+"Thread ID:"+Thread.currentThread().getName());
    			try {
    				Thread.sleep(1000);
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    		
    	}
    	public static void main(String[] args) {
    		MyTask task=new MyTask();
    		ExecutorService es=Executors.newFixedThreadPool(5);
    		for(int i=0;i<10;i++) {
    			es.submit(task);
    		}
    	}
    
    }

ExecutorService es=Executors.newFixedThreadPool(10);这行代码表示我创建了固定大小的线程池并且里面有10个线程。然后通过submit提交任务。最后输出如：

![70 2][]

输出就是这10个线程的执行情况。可以看到前5个任务和后5个任务的执行时间正好相差一秒。这正好符合一个只有5个线程池的行为。如果改为newCachedThreadPool输出入下：

![70 3][]

计划任务

另一个值得注意的方法就是newScheduledThreadPool()它返回一个ScheduledExecutorService对象，可以根据时间需要对线程进行调度。它的主要方法如下：

public ScheduledFuture<?> schedule(Runnable command,long delay, TimeUnit unit);
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,long delay, TimeUnit unit);
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                      long initialDelay,
                                                      long period,
                                                      TimeUnit unit);
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                         long initialDelay,
                                                         long delay,
                                                         TimeUnit unit);

与其他几个线程池不同，ScheduledExecutorService并不一定会立即安排执行任务。它起到的是计划任务的作用。它会在指定的时间对任务进行调度。schedule()会在给定的时间对任务进行一次调度。方法scheduleAtFixedRate()和scheduleWithFixedDelay()会对任务进行周期性调度。但是两者有点区别。对于scheduleAtFixedRate来说任务调度频率是一定的，它是以上一个任务开始执行时间为起点，然后的period时间调度下一次任务。而scheduleWithFixedDelay则是等上一个任务结束后再经过delay时间进行任务调度。

这个任务每5秒执行一次，内部会执行1秒钟，调度周期是5秒所以每5秒就会执行一次。

package thread.pool;
    
    import java.util.Date;
    import java.util.concurrent.Executors;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.TimeUnit;
    
    public class ScheduledExecutorServiceDemo {
    	public static void main(String[] args) {
    		ScheduledExecutorService se = Executors.newScheduledThreadPool(10);
    		//如果任务没完成则不会进行调度
    		se.scheduleAtFixedRate(new Runnable() {
    
    			@Override
    			public void run() {
    				try {
    					Date date=new Date(System.currentTimeMillis());
    					System.out.println("进来时间"+date);
    					Thread.sleep(1000);
    					System.out.println("出去时间"+new Date(System.currentTimeMillis()));
    				} catch (InterruptedException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    			}
    
    		}, 0, 5,TimeUnit.SECONDS );
    	}
    
    }

![70 4][]

如果出现执行时间大于调度时间则可以看到：

![70 5][]

当上一个任务执行完毕后，下一个任务会立即调用。

如果改成scheduleWithFixedDelay（）方法：

se.scheduleWithFixedDelay(new Runnable() {
    
    			@Override
    			public void run() {
    				try {
    					Date date=new Date(System.currentTimeMillis());
    					System.out.println("进来时间"+date);
    					Thread.sleep(1000);
    					System.out.println("出去时间"+new Date(System.currentTimeMillis()));
    				} catch (InterruptedException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    				
    			}
    			
    		}, 0, 5,TimeUnit.SECONDS );

![70 6][]

可以看到它的时间调度是根据上一个任务结束后再加上调度的时间然后再进行对run方法的调度。需要注意的是如果任务本身抛出了异常那么后续的所有执行都会被中断，所以对定时调度的异常处理十分重要。

3.核心线程池的内部实现

对于核心的几个线程池，无论是newFixedThreadPool()还是newSingleThreadExecutor、newCachedThreadPool()方法，其内部实现都是ThreadPoolExecutor实现的：

public static ExecutorService newFixedThreadPool(int nThreads) {
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>());
        }
     public static ExecutorService newSingleThreadExecutor() {
            return new FinalizableDelegatedExecutorService
                (new ThreadPoolExecutor(1, 1,
                                        0L, TimeUnit.MILLISECONDS,
                                        new LinkedBlockingQueue<Runnable>()));
        }
      public static ExecutorService newCachedThreadPool() {
            return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                          60L, TimeUnit.SECONDS,
                                          new SynchronousQueue<Runnable>());
        }

可以看到他们都是ThreadPoolExecutor类的封装。

然后看下ThreadPoolExecutor类的构造函数：

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler)

函数参数含义如下：

corePoolSize:指定了线程池中的线程数量。

maximumPoolSize：指定了线程池中最大线程数量。

keepAliveTime：当线程池线程数量超过了corePoolSize时，多余的空闲线程的存活时间。（超过corePoolSize的空闲线程在多场时间内会被销毁）

unit：keepAliveTime的单位。

workQueue：任务队列，被提交但尚未被执行的任务。

threadFactory：线程工厂，用于创建线程，一般是默认。

handler：拒绝策略。当任务太多来不及处理，如何拒绝任务。

参数workQueue指被提交但未执行的任务队列，它是一个BlockingQueue接口的对象，仅用于存放Runnable对象。根据队列功能分类，在ThreadPoolExecutor的构造器中可以 使用以下几种BlockingQueue。

1.直接提交的队列：该功能由SynchronousQueue对象提供。SynchronousQueue是一个特殊的BlockingQueue，它没有容量，每一个插入操作都要等待一个相应的删除操作，删除操作也要等待对应的插入操作。如果使用SynchronousQueue提交的任务不会被真实的保存，而总是将新任务提交给线程执行，如果没有空闲线程则尝试创建新的进程，如果进程数量已经达到最大值则执行拒绝策略。因此使用SynchronousQueue队列通常需要设置很大的maximumPoolSize值，否则很容易执行拒绝策略。

2.有界的任务队列：有界的任务队列可以使用ArrayBlockingQueue实现，ArrayBlockingQueue的构造器需要带一个容量仓鼠表示该队列的最大容量。public ArrayBlockingQueue(int capacity)。当使用有界的任务队列时，若有新的任务需要执行，如果线程池的实际线程数小于corePoolSize，则会优先创建新的线程，若大于corePoolSize，则会将新的任务假如等待队列。若等待队列已满，无法加入，则在总线程数不大于maximumPoolSize的前提下创建新的进程执行任务。若大于maximumPoolSize则执行拒绝策略。（除非系统非常繁忙，否则确保核心线程数维持在corePoolSize。）

3.无界的任务队列：无界的任务队列可以通过LinkedBlockingQueue类实现。与有界队列相比，除非系统资源耗尽，否则无界的任务队列不存在任务入队失败的情况。当有新的任务时，系统的线程小于corePoolSize时，线程池会生成新的线程执行任务，但当系统的线程数达到corePoolSize后就不会继续增加。如果任然有新的任务假如，而又没空闲的线程资源，则任务进入队列等待。如果任务创建和处理速度差异很大，无界队列会不断增长，直到耗尽系统内存。

4.优先任务队列：优先任务队列是带有执行优先级的队列。它通过PriorityBlockingQueue实现，可以控制任务的执行先后顺序。它是一个特殊的无界队列。无论是有界队列ArrayBlockingQueue还是未指定大小的无界队列LinkedBlockingQueue都是按照先进先出算法处理任务的。而PriorityBlockingQueue则根据任务自身的优先级顺序先后执行，取保系统性能的同时，也有很好的质量保证。

通过分析newFixedThreadPool()方法的实现可以发现corePoolSize和maximumPoolSize的大小是一样的，因为它是一个固定大小的线程池，并且对于固定大小的线程池而言，线程的数量不是动态变化的，所以corePoolSize和maximumPoolSize是相同的。而队列使用了LinkedBlockingQueue做任务队列。需要注意的是它使用无界队列存放无法立即执行的任务，当任务提交非常频繁时，该队列可能迅速膨胀从而耗尽系统资源。

newSingleThreadExecutor()方法返回的单线程线程池是newFixedThreadPool()方法的一种退化：其内部只是将corePoolSize和maximumPoolSize设为1.

newCachedThreadPool()方法则是corePoolSize设为0和maximumPoolSize设为无穷大。这意味着当没有任务时，这个线程池内无线程，而任务被提交时，该线程池会使用空闲的线程执行任务 ，若无空闲线程时，则将任务加入SynchronousQueue队列，而SynchronousQueue作为直接提交的队列，它会迫使线程池增加新的线程执行任务。当任务执行完毕后由于corePoolSize为0，所以空闲线程会在指定的时间内被回收。这里是60秒。需要注意的是，如果同时又大量任务被提交，而任务执行不快的话，那么系统会开启等量的线程处理，很有可能会很快耗尽系统资源。

ThreadPoolExecutor线程池调度的核心代码：

public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
            /*
             * Proceed in 3 steps:
             *
             * 1. If fewer than corePoolSize threads are running, try to
             * start a new thread with the given command as its first
             * task.  The call to addWorker atomically checks runState and
             * workerCount, and so prevents false alarms that would add
             * threads when it shouldn't, by returning false.
             *
             * 2. If a task can be successfully queued, then we still need
             * to double-check whether we should have added a thread
             * (because existing ones died since last checking) or that
             * the pool shut down since entry into this method. So we
             * recheck state and if necessary roll back the enqueuing if
             * stopped, or start a new thread if there are none.
             *
             * 3. If we cannot queue task, then we try to add a new
             * thread.  If it fails, we know we are shut down or saturated
             * and so reject the task.
             */
            int c = ctl.get();
            if (workerCountOf(c) < corePoolSize) {
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            if (isRunning(c) && workQueue.offer(command)) {
                int recheck = ctl.get();
                if (! isRunning(recheck) && remove(command))
                    reject(command);
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
            else if (!addWorker(command, false))
                reject(command);
        }

workerCountOf()函数回得到当前线程池的线程总数，当线程总数小于corePoolSize时会将任务通过addWorker()直接调度。否则会在workQueue.offer()处进入等待队列。如果进入等待队列失败（如有界队列达到上限，或者使用了SynchronousQueue）则会执行下一个的addWorker()将任务直接提交给线程池。如果当前线程数已经达到了maximumPoolSize则提交失败执行拒绝策略reject()。

![70 7][]

拒绝策略

ThreadPoolExecutor的最后一个参数指定了拒绝策略。拒绝策略是系统超负荷运行的补救措施，通常是压力太大引起的，也就是线程池中的线程已经用完了，无法继续为新任务服务，同时等待队列中也已经排满了，再也塞不下新任务，这时候就要用到拒绝策略。

JDK内置了四种拒绝策略：

![70 8][]

1.AbortPolicy策略：该策略会直接抛出异常，阻止系统正常工作。

2.CallerRunsPolicy策略：只要线程池未关闭，该策略直接在调用者线程中，运行当前被丢弃的任务。（这样做并不会真的丢弃任务，但是任务提交的性能可能会急剧下降。）

3.DiscardOledEstPolicy策略：该策略会丢弃最老的一个请求，也就是即将被执行的一个任务，并尝试再次提交当前任务。

4.DiscardPolicy策略：该策略会默默地丢弃无法处理的任务不做任何处理。（如果运行任务丢失，此方案最好。）

这几种内置策略均实现了RejectedExecutionHandler接口。

void rejectedExecution(Runnable r, ThreadPoolExecutor executor);

r表示请求执行的任务，executor为当前的线程池。

package thread.pool;
    
    import java.util.Date;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingDeque;
    import java.util.concurrent.RejectedExecutionHandler;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class RejectThreadPoolDemo {
        public static class MyTask implements Runnable {
            @Override
            public void run() {
                System.out.println(new Date(System.currentTimeMillis())
                        + ":Thread ID是:" + Thread.currentThread().getId());
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String args[]) throws InterruptedException {
            MyTask myTask = new MyTask();
    
            ExecutorService executorService = new ThreadPoolExecutor(5, 5, 0L,
                TimeUnit.SECONDS, 
                new LinkedBlockingDeque<Runnable>(10),   
                Executors.defaultThreadFactory()
                    , new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    //打印丢弃的任务
                    System.out.println("丢弃的任务是："+r.toString()+"线程池大小："+executor.getPoolSize());
                }
            });
            for (int i = 0; i < 100; i++) {
                executorService.submit(myTask);
                Thread.sleep(10);
            }
        }
    }

输出如下：

![70 9][]

这里我自定义了一个线程池，该线程池有5个线程，并且最大线程数为5。这和固定大小的线程池一样，但是它有一个只有10个容量的等待队列。因为使用无界队列时，如果任务量大可能会把内存撑爆。并且这里自定义了拒绝策略，我会打印出任务丢弃的信息。

![70 10][]

自定义线程创建：ThreadFactory

线程池中的线程由ThreadFactory得来。ThreadFactory是一个接口，它有只一个方法用来创建线程：

Thread newThread(Runnable r);

当线程池需要新建线程时，就会调用这个方法。

自定义线程可以自定义线程的名称，优先级等等，这里我将其设置为后台线程，并且打印了它们的ID。

ExecutorService executorService = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.SECONDS,
    				new LinkedBlockingDeque<Runnable>(10), new ThreadFactory() {
    
    					@Override
    					public Thread newThread(Runnable r) {
    						Thread t = new Thread(r);
    						t.setDaemon(true);
    						System.out.println("创建后台线程" + t+"ID:"+t.getId());
    						return t;
    					}
    				});
    		for (int i = 0; i < 10; i++) {
    			executorService.submit(myTask);
    		}
    		Thread.sleep(2000);

![70 11][]

扩展线程池

ThreadPoolExecutor是一个可扩展的线程池。它提供了beforeExecute()，afterExecute()，terminated()三个接口对线程池进行控制。

由于我这是1.8所以我的Worker调用的不是runTask()方法而是

final void runWorker(Worker w) {
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try {
                while (task != null || (task = getTask()) != null) {
                    w.lock();
                    // If pool is stopping, ensure thread is interrupted;
                    // if not, ensure thread is not interrupted.  This
                    // requires a recheck in second case to deal with
                    // shutdownNow race while clearing interrupt
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() &&
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                        wt.interrupt();
                    try {
                        beforeExecute(wt, task);
                        Throwable thrown = null;
                        try {
                            task.run();
                        } catch (RuntimeException x) {
                            thrown = x; throw x;
                        } catch (Error x) {
                            thrown = x; throw x;
                        } catch (Throwable x) {
                            thrown = x; throw new Error(x);
                        } finally {
                            afterExecute(task, thrown);
                        }
                    } finally {
                        task = null;
                        w.completedTasks++;
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally {
                processWorkerExit(w, completedAbruptly);
            }
        }

![70 12][]

而在默认的ThreadPoolExecutor实现中提供了空的beforeExecute()和afterExecute()实现。

下面来演示对线程池的扩展：

package thread.pool;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class ExtThreadPool {
    	public static class MyTask implements Runnable {
    		public String name;
    
    		public MyTask(String name) {
    			this.name = name;
    		}
    
    		@Override
    		public void run() {
    			System.out.println("正在执行:Thread ID:" + Thread.currentThread().getId() + ",Task Name:" + name);
    
    			try {
    				Thread.sleep(100);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    
    	public static void main(String args[]) throws InterruptedException {
    		ExecutorService executorService = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
    				new LinkedBlockingQueue<Runnable>()) {
    			protected void beforeExecute(Thread t, Runnable r) {
    				System.out.println("准备执行：" + ((MyTask) r).name);
    			}
    
    			protected void afterExecute(Runnable r, Throwable t) {
    				System.out.println("执行完成!!!!!!!!!" + ((MyTask) r).name);
    			}
    
    			protected void terminated() {
    				System.out.println("线程池退出！");
    			}
    		};
    
    		for (int i = 0; i < 5; i++) {
    			MyTask task = new MyTask("线程名-" + i);
    			executorService.execute(task);
    			Thread.sleep(10);
    		}
    		executorService.shutdown();
    	}
    }

![70 13][]

通过实现new LinkedBlockingQueue<Runnable>())这个接口我重写了beforeExecute()，afterExecute()和terminiated()三个方法。这三个方法方法分别展示了一个任务准备执行，任务执行结束，和整个线程池退出。

最后通过shutdown()方法关闭线程池。它会等待所有任务执行完成后再关闭线程池，但是它不会等待所有线程执行完毕后再返回。因此可以理解为shutdown()会发送一个关闭信号，在shutdown方法执行后这个线程池就不能再接受新的任务了。

![70 14][]

优化线程池线程数量

![70 15][]

**估算公式：Nthreads=Ncpu\*Ucpu\*(1+W/C)**

在java中可以通过Runtime.getRuntime().availableProcessors()获取CPU数量

参考《实战Java高并发程序设计》

[70]: /images/20220517/f3bfb4ecb39d439894ef4fe80e2aa715.png
[70 1]: /images/20220517/9b45e18342b24a9c90d052e0c2a3c410.png
[70 2]: /images/20220517/b94ea2c65ff04599b98ca6d3037ed870.png
[70 3]: /images/20220517/e62ec44fd4c7465db0c8601685b181e7.png
[70 4]: /images/20220517/3a00fbabad944cfb90a96536af5662a4.png
[70 5]: /images/20220517/a1962e849012447a958b2a1e19f4d496.png
[70 6]: /images/20220517/275aa5924b3e432db5362cbf120ea2cc.png
[70 7]: /images/20220517/0faa361e79c24ec79e2265cb1aee7525.png
[70 8]: /images/20220517/47a64039154a494d929b3ae23c6b6bbb.png
[70 9]: /images/20220517/881a08726f174d91b5c5d887b4b18d87.png
[70 10]: /images/20220517/f6ba748fb92d4b9caa31b0e42789f215.png
[70 11]: /images/20220517/6c9e4967fb7f4336bdefa0684342b13c.png
[70 12]: /images/20220517/b2f3e7a5cac146eaba138be6a9766d65.png
[70 13]: /images/20220517/0b71a1ebded34cab8ccf53a3af28ad40.png
[70 14]: /images/20220517/d2c436bc3ac8484e8b38cdd21cf9bbf0.png
[70 15]: /images/20220517/6a378312e1434b10b2b135aec701ce15.png