Java线程池核心原理剖析_JAVA_编程开发_程序员俱乐部

在系统开发时，我们经常会遇到“池”的概念。使用池一种以空间换时间的做法，通常在内存中事先保存一系列整装待命的对象，以供后期供其他对象随时调用。常见的池有：数据库连接池，socket连接池，线程池等。今天我们就来看一下线程池的概念。

◆
Executor
◆

JDK为我们提供了一套Executor框架来方便我们来管理和使用线程池。
打开java.util.concurrent.Executors类，我们可以发现JDK为我们提供了那么多的方法来帮助我们高效快捷的创建线程池：

class="line">1
2
3
4
5
6

public static ExecutorService newFixedThreadPool(int nThreads);//创建一个固定数目的、可重用的线程池
public static ExecutorService newSingleThreadExecutor();//创建一个单线程化的线程
public static ExecutorService newCachedThreadPool();//创建一个可缓存线程池
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize);//创建一个支持定时及周期性任务执行的线程池
public static ScheduledExecutorService newSingleThreadScheduledExecutor() ;//创建一个支持定时及周期性任务执行的线程池
public static ExecutorService newWorkStealingPool() ;//创建一个拥有多个任务队列的线程池

上方简单列举了几个Executor框架为我们提供的创建线程池的方法，这些线程池拥有各种各样的功能，我想当你刚刚开始使用线程的时候google如何使用线程池的时候大部分文章都是教你如何使用上方的一些方法创建一个线程池。但是如果你去查看他们的源码就会发现他们最后构造的时候都调用了同一个构造方法。（除了newWorkStealingPool之外，这个我们在下篇文章再讨论）

1
2
3
4
5
6
7

ThreadPoolExecutor(int corePoolSize,//线程池线程数量
                           int maximumPoolSize,//线程中最大的线程数量
                           long keepAliveTime,//线程池线程数量超过corePoolSize的空闲线程的存活时间
                           TimeUnit unit,//keepAliveTime时间单位
                           BlockingQueue<Runnable> workQueue,//被提交还没执行的任务存放在这里
                           ThreadFactory threadFactory,//线程工厂
                           RejectedExecutionHandler handler)//任务过多时的拒绝策略

上方的4个参数我想你看到了就会明白了，现在我们着重来讲一下下面的三个参数。

◆
WorkQueue
◆

参数workQueue是用来存放已提交但还未执行的任务，JDK为我们提供了一下实现：

直接提交队列SynchronousQueue

1
2
3
4
5
6
7
8
9
10

当新任务过来的时候它是这样处理的：
if（有空闲线程）{
    处理
}else{
    if(当前线程数<maximumPoolSize){
        创建新线程处理
    }else{
        执行拒绝策略
    }
}

因此使用这个队列时一定要设置很大的maximumPoolSize

有界的任务队列ArrayBlockingQueue

1
2
3
4
5
6
7
8
9
10
11
12
13

if（当前线程数<corePoolSize）{
    创建新线程执行
}else{
    if(任务队列是否已满){
       if(当前线程<maximumPoolSize){
          创建新线程处理
       }else{
          执行拒绝策略
        }
    }else{
       放到任务队列
    }
}

无界的任务队列LinkedBlockingDeque

1
2
3
4
5
6
7
8
9
10
11
12
13

if（当前线程数<corePoolSize）{
    创建新线程执行
}else{
    放入任务队列，等待执行，直到系统资源耗尽
}

优先任务队列PriorityBlockingQueue
根据任务的优先级将任务存放在任务队列特定位置
if（当前线程数<corePoolSize）{
    创建新线程执行
}else{
    等待执行，直到系统资源耗尽
}

◆
线程工厂
◆

第六个参数threadFactory是为线程池中创建线程的，我们使用Executor框架创建的线程就是有threadFactory提供的。我们看一下JDK提供的默认的threadFactory：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

重点关注一下其中的newThread方法，看到这个我想你就明白了为什么你使用线程池创建出来的线程打印的时候名字的来源，还有是否是守护线程和优先级等属性的来源了。

◆
拒绝策略
◆

看到刚刚的几种任务队列我们发现当任务过多时是需要指定拒绝策略来进行拒绝呢，那么JDK又为我们提供了哪些拒绝策略呢。

1
2
3
4

AbortPolicy直接抛出异常。
CallerRunsPolicy：如果线程池未关闭，则在调用者线程中运行当前任务
DiscardOldestPolicy：丢弃即将执行的任务，然后再尝试提交当前任务
DiscardPolicy：丢弃此任务

◆
线程池的扩展
◆

ThreadPoolExecutor不仅仅能够创建各种各样的线程来帮助我们实行功能，它还预留了三个接口来供我们进行扩展。

在runWorker方法中调用线程进行执行之前调用了beforeExecute方法，执行之后调用了afterExecute()方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

final void runWorker(Worker w) {
       Thread wt = Thread.currentThread();
       Runnable task = w.firstTask;
       w.firstTask = null;
       w.unlock(); 
       boolean completedAbruptly = true;
       try {
           while (task != null || (task = getTask()) != null) {
               w.lock();
               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);
       }
   }

这两个方法在ThreadPoolExecutor类中是没有实现的，我们想要监控线程运行前后的数据就可以通过继承ThreadPoolExecutor类来实现这个扩展。
另外还有一个terminated()方法是在整个线程池退出的时候调用的，我们这里一并扩展。

public class ThreadPoolExecutorDemo extends ThreadPoolExecutor {
    //注意这里因为ThreadPoolExecutor没有无参的构造，所以还需要重写一下构造方法。
    //这里限于篇幅就不贴了
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        System.out.println(Thread.currentThread().getId()+"执行完成");

    }
    @Override
    protected void terminated() {
        System.out.println("线程池退出");
    }
}

//使用这个demo就可以验证我们扩展的结果了。

public class ThreadPoolDemo {
    static class ThreadDemo extends Thread {
        @Override
        public void run() {
            System.out.println(System.currentTimeMillis() + ":Thread ID is:" + Thread.currentThread().getId());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutorDemo threadPoolExecutorDemo=  new ThreadPoolExecutorDemo(5,5,0,TimeUnit.SECONDS,new LinkedBlockingDeque<Runnable>());
        ThreadDemo threadDemo = new ThreadDemo();
        for (int i = 0; i < 20; i++) {
            threadPoolExecutorDemo.submit(threadDemo);
        }
        threadPoolExecutorDemo.shutdown();
    }
}

本文所有源码：https://github.com/shiyujun/syj-study-demo

?

• 
• 大小: 171.6 KB

• 查看图片附件