[Java知识库]Java之线程池的原理及实现

Java之线程池的原理及实现

什么是线程池？

Java中的线程池是运用场景最多的并发框架，几乎所有需要异步或并发执行任务的程序
都可以使用线程池。在开发过程中，合理地使用线程池能够带来3个好处。
第一：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁所造成的消耗。
第二：提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
第三：提高线程的可管理性。线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性。使用线程池可以进行统一分配，调优和监控。但是，要合理利用线程池，必须对其实现原理了如指掌。

例如 ：假设一个服务器完成一项任务所需时间为：T1 创建线程时间，T2 在线程中执行任务的时间，T3 销毁线程时间。

如果：T1 + T3 远大于 T2，则可以采用线程池，以提高服务器性能。
也就是说，如果我去医院看医生，我去的时候医生已经在这里了，而不是我去的时候医生才开始学医，大大节约了创建线程和销毁线程的重复的操作。

线程池的作用

线程池是为突然大量爆发的线程设计的，通过有限的几个固定线程为大量的操作服务，减少了创建和销毁线程所需的时间，从而提高效率。
如果一个线程的时间非常长，就没必要用线程池了(不是不能作长时间操作，而是不宜。)，况且我们还不能控制线程池中线程的开始、挂起、和中止。

线程池的四种创建方式

Java通过Executors（jdk1.5并发包）提供四种线程池，分别为：

newCachedThreadPoo：l创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

public void Demo(){
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            int a = i;
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName()+ "  :"+a);

                }
            });
        }
    }

newFixedThreadPool： 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

public void Demo(){
        ExecutorService executorService = Executors.newFixedThreadPool(3);//3为最大并发数
        for (int i = 0; i < 20; i++) {
            int a= i;
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+ "  :"+a);
                }
            });
        }
    }

newScheduledThreadPool： 创建一个定长线程池，支持定时及周期性任务执行。

public void Demo(){
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);//同方法二 定长
        for (int i = 0; i < 20; i++) {
            int a= i;
            executorService.schedule(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+ "  :"+a);
                }
            },3,TimeUnit.SECONDS); //设置定时
        }

    }

newSingleThreadExecutor： 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

public void singleDemo(){
        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
        for (int i = 0; i < 10; i++) {
            int a= i;
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+ "  :"+a);
                }
            });
        }

    }

这是我们Java自带的四种建立线程池的方式

线程池的七大参数

1.corePoolSize核心线程数量 线程池长期存在的几个线程数量,对任务进行处理
2.maximunPoolSize 最大线程数,当核心线程全部在工作之后,再来任务,进入阻塞队列中,等待执行,在阻塞队列满了之后,就打开新的线程,直到线程数量达到最大线程数量
3.KeepAliveTime 空闲线程存活时间,当多个线程处理任务,线程数量大于核心线程数量,此时又有线程处于空闲状态的时候,此时,将空闲线程销毁,保持核心线程数量就可以了
4.unit存活时间单位
5.workQuene阻塞队列,见上面
6.threadFactory 线程工厂,常用默认DefaultThreadFactory()
7.拒绝策略：在使用线程池并且使用有界队列的时候，如果队列满了，任务添加到线程池的时候就会有问题
拒绝策略有四种：

        第一种AbortPolicy:不执行新任务，直接抛出异常，提示线程池已满

         第二种DisCardPolicy:不执行新任务，也不抛出异常

         第三种DisCardOldSetPolicy:将消息队列中的第一个任务替换为当前新进来的任务执行

         第四种CallerRunsPolicy:直接调用execute来执行当前任务

线程池的原理

上面提到了线程池的七大参数，其中提到了等待队列，在线程池的核心原理上，等待队列起到了一个非常大的作用。

主要流程图可以画为：

我们在往线程池添加线程进行执行的时候，当线程池中没有正在执行的线程，那么该线程直接进入核心线程区进行执行，当核心线程区满了，就进如等待队列，同时让该线程wait（）；等待核心线程区有空间时，当等待队列中存满了，在进入的线程就会进入非核心进程区，最后装不下的就进行拒绝。

public class MyThreadPool extends ThreadGroup{
    private boolean isClosed = false;
    private LinkedList queue;
    private static int threadPool_ID = 1;

    public MyThreadPool(int poolSize) {
        super(threadPool_ID+"");
        setDaemon(true);
        queue = new LinkedList();
        for (int i = 0;i<poolSize;i++){
            new MyThread(i).start();
        }
    }

    class MyThread extends Thread{
        private int id;
        public MyThread(int id){
            super(MyThreadPool.this,id+"");
            this.id = id;
        }

        @Override
        public void run() {
            while (!isInterrupted()){
                Runnable task = null;

                try {
                    task = getTask(id);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (task == null){
                    return;
                }
                task.run();
            }
        }
    }
    private synchronized Runnable getTask(int id) throws InterruptedException {
        while (queue.size()==0){
            if (isClosed){
                return null;
            }
            System.out.println("工作线程"+id+"等待任务");
            wait();
        }
        System.out.println("工作线程"+id+"开始执行任务");

        return (Runnable) queue.removeFirst();
    }
    public synchronized void addThread(Runnable task){
        if (isClosed){
            return;
        }
        if(task !=null){
            queue.add(task);// 向任务队列中添加一个任务
            notify();//唤醒
        }
    }
    public synchronized void closed(){
        if (!isClosed){
            isClosed = true;
            queue.clear();
            interrupt();
        }
    }
    public void on() throws InterruptedException {
        synchronized (this){
            isClosed = true;
            notifyAll();
        }
        Thread [] threads = new Thread[activeCount()];
        int count = enumerate(threads);
        for (int i = 0;i<count;i++){
            threads[i].join();
        }
    }

}