【＜Java中的线程池＞】

我就是我 2023-03-06 05:05 24阅读 0赞

### Java中的线程池 ###

*   *   *   *   *  \[1\] 使用线程池的好处
                 *  \[2\] Executor框架
                 *  \[3\] ThreadPoolExecutor类的参数字段
                 *  \[4\] 线程池的排队策略
                 *  \[6\] 拒绝策略
                 *  \[5\] 常见的阻塞队列
                 *  \[6\] Java提供的四种线程池
                 *  \[7\] 手写一个线程池

##### \[1\] 使用线程池的好处 #####

为了降低资源消耗，提高响应速度，提高线程的可管理性，所以出现了线程池。

1.  **降低资源消耗**：通过重复利用已创建的线程降低线程创建和销毁造成的消耗;
2.  **提高响应速度**：当任务到达时，任务可以不需要的等到线程创建就能立即执行;
3.  **提高线程的可管理性**：线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

##### \[2\] Executor框架 #####

https://www.cnblogs.com/zxrxzw/p/12116209.html)  
**1. 什么是Executor框架？**

我们知道线程池就是线程的集合，线程池集中管理线程，以实现线程的重用，降低资源消耗，提高响应速度等。线程用于执行异步任务，单个的线程既是工作单元也是执行机制。从JDK1.5开始，为了把工作单元与执行机制分离开，**Executor框架是一个统一创建与运行的接口。Executor框架实现的就是线程池的功能。**

**2. Executor框架结构图解**

Executor框架包括3大部分：

（1）**任务**：也就是工作单元，包括被执行任务需要实现的接口：Runnable接口或者Callable接口；

（2）**任务的执行**：也就是把任务分派给多个线程的执行机制，包括Executor接口及继承自Executor接口的ExecutorService接口。

（3）**异步计算的结果**：包括Future接口及实现了Future接口的[FutureTask][]类。

**3. Executor框架成员：**

ThreadPoolExecutor实现类、ScheduledThreadPoolExecutor实现类、Future接口、Runnable和Callable接口、Executors工厂类

![img][]

##### \[3\] ThreadPoolExecutor类的参数字段 #####

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

*  **corePoolSize：核心线程数**：线程池中会维护一个最小的线程数量，即使这些线程处于空闲状态，他们也不会被销毁（除非设置了allowCoreThreadTimeOut），而且线程数小于此值的话，仍然会创建新的线程。
 *  **maximumPoolSize：最大线程数**：线程池所能容的最多线程数量。一个任务被提交到线程池后，首先会缓存到工作队列中，如果工作队列满了，则会创建一个新线程，然后从工作队列中的取出一个任务交由新线程来处理，而将刚提交的任务放入工作队列。线程池不会无限制的去创建新线程，它会有一个最大线程数量的限制，这个数量即由maximunPoolSize来指定。
 *  **keepAliveTime ：线程空闲的保活时间**：一个线程如果处于空闲状态，并且当前的线程数量大于corePoolSize，那么在指定时间后，这个空闲线程会被销毁，这里的指定时间由keepAliveTime来设定。
 *  **unit：时间单位**：keepAliveTime的计量单位。
 *  **workQueue：存储线程的队列**：新任务被提交后，会先进入到此工作队列中，任务调度时再从队列中取出任务。jdk中提供了四种工作队列。
 *  **threadFactory：创建线程的工厂**：创建一个新线程时使用的工厂，可以用来设定线程名、是否为daemon线程等等
 *  **handler：拒绝策略**：当工作队列中的任务已到达最大限制，并且线程池中的线程数量也达到最大限制，这时如果有新任务提交进来，该如何处理呢。这里的拒绝策略，就是解决这个问题的，jdk中提供了4中拒绝策略。

##### \[4\] 线程池的排队策略 #####

1.  当线程数达到 小于corePoolSize ，即使有空闲队列，线程池也会优先创建一个新线程来执行任务。
2.  当线程数达到 corePoolSize 且并没有线程空闲时，加入任务，这时候会有两种情况：
    
     *  如果队列选择了**无界队列**，新加的任务会被加入workQueue 队列排队，直到有空闲的线程。
     *  如果队列选择了**有界队列**，任务会优先入队，那么任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线程来救急，若此范围线程空闲，则有keepAliveTime决定存活时间。
3.  **针对有界队列**，如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 4 种实现，其它著名框架也提供了实现。
4.  **针对有界队列**，当高峰过去后，超过corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime 和 unit 来控制。

##### \[6\] 拒绝策略 #####

**四种线程池拒绝策略**

当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize时，如果还有任务到来就会采取任务拒绝策略，通常有以下四种策略：

*  ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
 *  ThreadPoolExecutor.DiscardPolicy：丢弃任务，但是不抛出异常。
 *  ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新提交被拒绝的任务
 *  ThreadPoolExecutor.CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务

**框架实现：**

*  Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方  
    便定位问题
 *  Netty 的实现，是创建一个新线程来执行任务
 *  ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略
 *  PinPoint 的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略

##### \[5\] 常见的阻塞队列 #####

https://blog.csdn.net/xiewenfeng520/article/details/107142566

前面我们介绍了线程池内部有一个排队策略，任务可能需要在队列中进行排队等候。常见的阻塞队列包括如下的四种，接下来我们一起来看看吧。

*  **ArrayBlockingQueue**：**基于数组的有界阻塞队列，按FIFO排序**。新任务进来后，会放到该队列的队尾，有界的数组可以防止资源耗尽问题。当线程池中线程数量达到corePoolSize后，再有新任务进来，则会将任务放入该队列的队尾，等待被调度。如果队列已经是满的，则创建一个新线程，如果线程数量已经达到maxPoolSize，则会执行拒绝策略。
 *  **LinkedBlockingQuene**：**基于链表的无界阻塞队列（其实最大容量为Interger.MAX），按照FIFO排序**。由于该队列的近似无界性，当线程池中线程数量达到corePoolSize后，再有新任务进来，会一直存入该队列，而不会去创建新线程直到maxPoolSize，因此使用该工作队列时，参数maxPoolSize其实是不起作用的。
 *  **SynchronousQuene**：**一个不缓存任务的阻塞队列，生产者放入一个任务必须等到消费者取出这个任务**。也就是说新任务进来时，不会缓存，而是直接被调度执行该任务，如果没有可用线程，则创建新线程，如果线程数量达到maxPoolSize，则执行拒绝策略。
 *  **PriorityBlockingQueue**：**具有优先级的无界阻塞队列**，优先级通过参数Comparator实现。

> **阻塞队列特点及其使用场景：**
> 
> **ArrayBlockingQueue（有界）:**
> 
>  *  内部使用一个**数组**作为其存储空间，数组的存储空间是**预先分配**的
>  *  **优点是** put 和 take操作不会增加GC的负担（因为空间是预先分配的）
>  *  **缺点是** put 和 take操作使用同一个锁，可能导致锁争用，导致较多的上下文切换。
>  *  会执行拒绝策略，有救急线程
>  *  ArrayBlockingQueue适合在生产者线程和消费者线程之间的**并发程度较低**的情况下使用。
> 
> **LinkedBlockingQueue：**
> 
>  *  是一个无界队列（其实队列长度是Integer.MAX\_VALUE）
>  *  内部存储空间是一个**链表**，并且链表节点所需的**存储空间是动态分配**的
>  *  **优点是** put 和 take 操作使用两个显式锁（putLock和takeLock）
>  *  **缺点是**增加了GC的负担，因为空间是动态分配的。
>  *  不会执行拒绝策略，无救急线程
>  *  LinkedBlockingQueue适合在生产者线程和消费者线程之间的**并发程度较高**的情况下使用。
> 
> **SynchronousQueue：**
> 
>  *  SynchronousQueue可以被看做一种特殊的**有界**队列。
>  *  生产者线程生产一个产品之后，会等待消费者线程来取走这个产品，才会接着生产下一个产品。
>  *  适合在生产者线程和消费者线程之间的处理**能力相差不大**的情况下使用。

##### \[6\] Java提供的四种线程池 #####

Executors只是对线程池一些特定情况的简洁使用，直接用ThreadPoolExecutor构造线程池会获得更为强大的功能。java自带的四种线程池通过 ThreadPoolExecutor 的构造方法实现：

1.  **newCachedThreadPool**：是一个根据需要来创建线程的线程池，它的corePoolSize为0，maximumPoolSize设置为Integer.MAX\_VALUE，所以maximumPool是**无界**的，keepAliveTime设置为60L，代表所有空闲线程等待新任务的最长时间为60L。

> 这意味着，如果主线程提交任务的速度高于maxiunmPool中线程处理任务的速度时，将会不断CachedThreadPool创建新线程。极端情况下会耗尽CPU资源。所以它适用于**负载较轻**的场景，执行短期异步任务如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。（可以使得任务快速得到执行，因为任务时间执行短，可以很快结束，也不会造成cpu过度切换）

public static ExecutorService newCachedThreadPool() { 
            return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                          60L, TimeUnit.SECONDS,
                                          new SynchronousQueue<Runnable>());
        }

1.  **newFixedThreadPool**：创建一个**可重用固定线程数**的线程池，可控制线程最大并发数，超出的线程会在队列中等待。因为采用无界的阻塞队列LinkedBlockingQueue，所以maximumPoolSize和keepAliveTime为无效函数，实际线程数量永远不会变化，只会小于等于corePoolSize。

> 适用于**负载较重**的场景，对当前线程数量进行限制。（保证线程数可控，不会造成线程过多，导致系统负载更为严重）

public static ExecutorService newFixedThreadPool(int nThreads) { 
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>());
    }

1.  **newSingleThreadExecutor**：创建一个**单线程**的线程池，适用于需要保证顺序执行各个任务。它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。它的corePoolSize和keepAliveTime设置为1，由于采用无界堵塞队列LinkedBlockingQueue，可以保证无限缓存任务，并且线程固定。

public static ExecutorService newSingleThreadExecutor() { 
            return new FinalizableDelegatedExecutorService
                (new ThreadPoolExecutor(1, 1,
                                        0L, TimeUnit.MILLISECONDS,
                                        new LinkedBlockingQueue<Runnable>()));
    }

1.  **newScheduledThreadPool**：它继承于ThreadPoolExecutor。创建一个**定长任务**线程池，适用于执行延时或者周期性任务。

public class OneMoreStudy { 
        public static void main(String[] args) { 
            final SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
            ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(3);
            System.out.println("提交时间: " + sdf.format(new Date()));
            scheduledThreadPool.schedule(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println("运行时间: " + sdf.format(new Date()));
                    }
                }, 3, TimeUnit.SECONDS);
            scheduledThreadPool.shutdown();
        }
    }

##### \[7\] 手写一个线程池 #####

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-8aejg5JF-1596598779485)(X:\\Users\\xu\\AppData\\Roaming\\Typora\\typora-user-images\\image-20200709145932351.png)\]

package cn.itcast.n8;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.core.log.LogDelegateFactory;
    
    import java.util.ArrayDeque;
    import java.util.Deque;
    import java.util.HashSet;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    
    @Slf4j(topic = "c.TestPool")
    public class TestPool { 
        public static void main(String[] args) { 
            ThreadPool threadPool = new ThreadPool(1,
                    1000, TimeUnit.MILLISECONDS, 1, (queue, task)->{ 
                // 1. 死等
    // queue.put(task);
                // 2) 带超时等待
    // queue.offer(task, 1500, TimeUnit.MILLISECONDS);
                // 3) 让调用者放弃任务执行
    // log.debug("放弃{}", task);
                // 4) 让调用者抛出异常
    // throw new RuntimeException("任务执行失败 " + task);
                // 5) 让调用者自己执行任务
                task.run();
            });
            for (int i = 0; i < 4; i++) { 
                int j = i;
                threadPool.execute(() -> { 
                    try { 
                        Thread.sleep(1000L);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    log.debug("{}", j);
                });
            }
        }
    }
    
    @FunctionalInterface // 拒绝策略
    interface RejectPolicy<T> { 
        void reject(BlockingQueue<T> queue, T task);
    }
    
    @Slf4j(topic = "c.ThreadPool")
    class ThreadPool { 
        // 任务队列
        private BlockingQueue<Runnable> taskQueue;
    
        // 线程集合
        private HashSet<Worker> workers = new HashSet<>();
    
        // 核心线程数
        private int coreSize;
    
        // 获取任务时的超时时间
        private long timeout;
    
        private TimeUnit timeUnit;
    
        private RejectPolicy<Runnable> rejectPolicy;
    
        // 执行任务
        public void execute(Runnable task) { 
            // 当任务数没有超过 coreSize 时，直接交给 worker 对象执行
            // 如果任务数超过 coreSize 时，加入任务队列暂存
            synchronized (workers) { 
                if(workers.size() < coreSize) { 
                    Worker worker = new Worker(task);
                    log.debug("新增 worker{}, {}", worker, task);
                    workers.add(worker);
                    worker.start();
                } else { 
    // taskQueue.put(task);
                    // 1) 死等
                    // 2) 带超时等待
                    // 3) 让调用者放弃任务执行
                    // 4) 让调用者抛出异常
                    // 5) 让调用者自己执行任务
                    taskQueue.tryPut(rejectPolicy, task);
                }
            }
        }
    
        public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity, RejectPolicy<Runnable> rejectPolicy) { 
            this.coreSize = coreSize;
            this.timeout = timeout;
            this.timeUnit = timeUnit;
            this.taskQueue = new BlockingQueue<>(queueCapcity);
            this.rejectPolicy = rejectPolicy;
        }
    
        class Worker extends Thread{ 
            private Runnable task;
    
            public Worker(Runnable task) { 
                this.task = task;
            }
    
            @Override
            public void run() { 
                // 执行任务
                // 1) 当 task 不为空，执行任务
                // 2) 当 task 执行完毕，再接着从任务队列获取任务并执行
    // while(task != null || (task = taskQueue.take()) != null) { 
                while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) { 
                    try { 
                        log.debug("正在执行...{}", task);
                        task.run();
                    } catch (Exception e) { 
                        e.printStackTrace();
                    } finally { 
                        task = null;
                    }
                }
                synchronized (workers) { 
                    log.debug("worker 被移除{}", this);
                    workers.remove(this);
                }
            }
        }
    }
    @Slf4j(topic = "c.BlockingQueue")
    class BlockingQueue<T> { 
        // 1. 任务队列
        private Deque<T> queue = new ArrayDeque<>();
    
        // 2. 锁
        private ReentrantLock lock = new ReentrantLock();
    
        // 3. 生产者条件变量
        private Condition fullWaitSet = lock.newCondition();
    
        // 4. 消费者条件变量
        private Condition emptyWaitSet = lock.newCondition();
    
        // 5. 容量
        private int capcity;
    
        public BlockingQueue(int capcity) { 
            this.capcity = capcity;
        }
    
        // 带超时阻塞获取
        public T poll(long timeout, TimeUnit unit) { 
            lock.lock();
            try { 
                // 将 timeout 统一转换为 纳秒
                long nanos = unit.toNanos(timeout);
                while (queue.isEmpty()) { 
                    try { 
                        // 返回值是剩余时间
                        if (nanos <= 0) { 
                            return null;
                        }
                        nanos = emptyWaitSet.awaitNanos(nanos);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                T t = queue.removeFirst();
                fullWaitSet.signal();
                return t;
            } finally { 
                lock.unlock();
            }
        }
    
        // 阻塞获取
        public T take() { 
            lock.lock();
            try { 
                while (queue.isEmpty()) { 
                    try { 
                        emptyWaitSet.await();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                T t = queue.removeFirst();
                fullWaitSet.signal();
                return t;
            } finally { 
                lock.unlock();
            }
        }
    
        // 阻塞添加
        public void put(T task) { 
            lock.lock();
            try { 
                while (queue.size() == capcity) { 
                    try { 
                        log.debug("等待加入任务队列 {} ...", task);
                        fullWaitSet.await();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                log.debug("加入任务队列 {}", task);
                queue.addLast(task);
                emptyWaitSet.signal();
            } finally { 
                lock.unlock();
            }
        }
    
        // 带超时时间阻塞添加
        public boolean offer(T task, long timeout, TimeUnit timeUnit) { 
            lock.lock();
            try { 
                long nanos = timeUnit.toNanos(timeout);
                while (queue.size() == capcity) { 
                    try { 
                        if(nanos <= 0) { 
                            return false;
                        }
                        log.debug("等待加入任务队列 {} ...", task);
                        nanos = fullWaitSet.awaitNanos(nanos);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                log.debug("加入任务队列 {}", task);
                queue.addLast(task);
                emptyWaitSet.signal();
                return true;
            } finally { 
                lock.unlock();
            }
        }
    
        public int size() { 
            lock.lock();
            try { 
                return queue.size();
            } finally { 
                lock.unlock();
            }
        }
    
        public void tryPut(RejectPolicy<T> rejectPolicy, T task) { 
            lock.lock();
            try { 
                // 判断队列是否满
                if(queue.size() == capcity) { 
                    rejectPolicy.reject(this, task);
                } else {   // 有空闲
                    log.debug("加入任务队列 {}", task);
                    queue.addLast(task);
                    emptyWaitSet.signal();
                }
            } finally { 
                lock.unlock();
            }
        }
    }

[FutureTask]: https://www.cnblogs.com/zhumiao/p/9492360.html
[img]: /images/20230208/5c11630ee70a40edae8097d30f8c6b24.png