jdk源码解析

1. jdk源码

大家可以多看源码，看一下同样的功能，代码是如何构造的；

• ThreadpoolExector
  

1.1线程池

我们通常所说的线程池是指Java中的ThreadPoolExecutor，下面将详细说明线程池的参数、实现原理以及如何实现一个简单的线程池。

线程池的参数（以ThreadPoolExecutor为例）

ThreadPoolExecutor有7个核心参数：

1. corePoolSize（核心线程数）：线程池中保持存活的最小线程数，即使它们处于空闲状态。
2. maximumPoolSize（最大线程数）：线程池中允许存在的最大线程数。
3. keepAliveTime（线程存活时间）：当线程数超过核心线程数时，多余的空闲线程在终止前等待新任务的最长时间。
4. unit（时间单位）：keepAliveTime的时间单位。
5. workQueue（工作队列）：用于保存等待执行的任务的阻塞队列。
6. threadFactory（线程工厂）：用于创建新线程的工厂。
7. handler（拒绝策略）：当线程池和工作队列都满了，说明线程池处于饱和状态，必须采取一种策略处理新提交的任务。

线程池的实现原理

线程池的主要工作流程如下：

1. 提交任务后，线程池首先判断当前线程数是否小于核心线程数（corePoolSize）。如果是，则创建一个新线程来执行任务。
2. 如果当前线程数已经达到核心线程数，则将任务放入工作队列（workQueue）中。
3. 如果工作队列已满，且当前线程数小于最大线程数（maximumPoolSize），则创建新的线程来执行任务。
4. 如果当前线程数已经达到最大线程数，并且工作队列已满，则按照拒绝策略（handler）处理新提交的任务。

线程池中的线程执行完任务后，会从工作队列中获取新的任务来执行。如果超过keepAliveTime仍然没有获取到新任务，并且当前线程数大于核心线程数，则该线程会被终止，直到线程数等于核心线程数。

拒绝策略详解

Java提供了4种内置拒绝策略：

1. AbortPolicy：直接抛出RejectedExecutionException（默认）

2. CallerRunsPolicy：由调用者线程执行该任务

3. DiscardPolicy：直接丢弃任务，不抛异常

4. DiscardOldestPolicy：丢弃队列中最老的任务，然后重试执行

5. FixedThreadPool（固定大小线程池）

6. CachedThreadPool（缓存线程池）

7. SingleThreadExecutor（单线程线程池）

8. ScheduledThreadPool（定时任务线程池）

9. WorkStealingPool（工作窃取线程池，Java 8+）

10. ForkJoinPool（分治线程池，与WorkStealingPool相关）

下面分别介绍这些线程池的特点和适用场景。

1. FixedThreadPool（固定大小线程池）

• 创建方法：Executors.newFixedThreadPool(int nThreads)
• 特点：核心线程数和最大线程数相等，即固定大小的线程池。线程池中的线程不会因为任务过多而扩容，也不会因为空闲而收缩。任务队列使用无界队列（LinkedBlockingQueue）。
• 适用场景：适用于负载较重的服务器，需要限制当前线程数量，以及需要稳定线程数的场景。

2. CachedThreadPool（缓存线程池）

• 创建方法：Executors.newCachedThreadPool()
• 特点：核心线程数为0，最大线程数为Integer.MAX_VALUE，线程空闲存活时间为60秒。任务队列使用同步队列（SynchronousQueue），该队列不存储元素，每个插入操作必须等待另一个线程的移除操作。因此，如果有空闲线程则复用，如果没有则创建新线程。线程池的线程数可以无限扩大，当线程空闲时会被回收。
• 适用场景：适用于执行很多短期异步任务，或者负载较轻的服务器。

3. SingleThreadExecutor（单线程线程池）

• 创建方法：Executors.newSingleThreadExecutor()
• 特点：只有一个线程的线程池，核心线程数和最大线程数均为1。任务队列使用无界队列（LinkedBlockingQueue）。保证所有任务按顺序执行。
• 适用场景：适用于需要顺序执行任务的场景，并且不需要并发执行。

4. ScheduledThreadPool（定时任务线程池）

• 创建方法：Executors.newScheduledThreadPool(int corePoolSize)
• 特点：可以执行定时任务和周期性任务。核心线程数由参数指定，最大线程数为Integer.MAX_VALUE，线程空闲存活时间为0，但实际上由于用于调度任务，所以不会回收核心线程。任务队列使用延迟工作队列（DelayedWorkQueue）。
• 适用场景：适用于需要执行定时任务和周期性任务的场景。

5. WorkStealingPool（工作窃取线程池）

• 创建方法：Executors.newWorkStealingPool(int parallelism)， parallelism并行级别，默认为CPU核数。
• 特点：基于ForkJoinPool实现，使用工作窃取算法，即每个线程维护自己的任务队列，当自己队列中的任务执行完后，可以从其他线程的任务队列中“窃取”任务来执行。这种机制可以减少线程间的竞争，提高吞吐量。
• 适用场景：适用于任务执行时间不均匀，或者有大量子任务的场景。

6. ForkJoinPool（分治线程池）

• 创建方法：new ForkJoinPool(int parallelism)
• 特点：与WorkStealingPool类似，也是基于工作窃取算法。ForkJoinPool是WorkStealingPool的底层实现，但WorkStealingPool是ForkJoinPool的封装，调整了一些默认参数。ForkJoinPool适合处理可以递归分解的任务（分治任务）。
• 适用场景：适用于计算密集型任务，且任务可以分解成更小的子任务。

注意事项

• FixedThreadPool和SingleThreadExecutor使用无界队列，如果任务提交速度大于处理速度，可能会堆积大量任务，导致内存溢出。
• CachedThreadPool和ScheduledThreadPool的最大线程数很大，如果任务提交过多，可能会创建大量线程，导致资源耗尽。

因此，在实际使用中，通常建议根据业务场景自定义线程池参数，使用ThreadPoolExecutor构造函数来创建，以便更精确地控制线程池的行为。

常见配置建议：

• CPU密集型：线程数 ≈ CPU核数
• IO密集型：线程数 ≈ CPU核数 × (1 + 平均等待时间/平均计算时间)
• 混合型：根据实际情况调整

//中断空闲的线程 private void interruptIdleWorkers(boolean onlyOne){final ReentrantLock mainLock=this.mainLock;//专用锁 mainLock.lock();//添加专用锁 try{for(Worker w:workers){Thread t=w.thread;if(!t.isInterrupted()&&w.tryLock()){try{t.interrupt();//阻塞当前线程}finally{w.unlock();}}if(onlyOne)break;}}finally{mainLock.unlock();//解除专用锁代码}}public booleantryLock(){returnthis.tryAcquire(1);}protected boolean tryAcquire(int unused){if(this.compareAndSetState(0,1)){//cas线程安全 this.setExclusiveOwnerThread(Thread.currentThread());returntrue;}else{returnfalse;}}public voidunlock(){this.release(1);}public final boolean release(int arg){if(tryRelease(arg)){signalNext(head);returntrue;}returnfalse;}protected boolean tryRelease(int unused){setExclusiveOwnerThread(null);setState(0);returntrue;}// setExclusiveOwnerThread(Thread.currentThread());线程非安全，但是由于有cas锁机制是线程安全的;protected boolean tryAcquire(int unused){if(compareAndSetState(0,1)){setExclusiveOwnerThread(Thread.currentThread());returntrue;}returnfalse;}protected boolean tryRelease(int unused){this.setExclusiveOwnerThread((Thread)null);this.setState(0);returntrue;}//若 tryAcquire 和 tryRelease分别两个专项锁，则可能出现线程安全的问题;设计规则上是 若 tryAcquire 和 tryRelease 一块使用 关键因素 ReentrantLock 机制保障 两个方法都在 ReentrantLock 的同步机制控制下 tryAcquire 和 tryRelease 方法的调用都受到同一把锁的保护 互斥访问 ReentrantLock 确保同一时刻只有一个线程能执行临界区代码 无论是获取锁还是释放锁，都是在受保护的上下文中进行 内存可见性 ReentrantLock 的获取和释放操作建立了 happens-before 关系 保证了 setExclusiveOwnerThread 操作的内存可见性 源码方法 private volatile ThreadFactory threadFactory;/** * Sets the thread factory used to create new threads. * * @param threadFactory the new thread factory * @throws NullPointerExceptionifthreadFactory is null * @see#getThreadFactory*/ public void setThreadFactory(ThreadFactory threadFactory){Objects.requireNonNull(threadFactory,"threadFactory");this.threadFactory=threadFactory;//线程安全单独的赋值操作}/** * Initiates an orderlyshutdowninwhichpreviously submitted * tasks are executed, but no new tasks will be accepted. * Invocation has no additional effectifalready shut down. * *<p>This method does notwaitforpreviously submitted tasks to * complete execution. Use{@link#awaitTermination awaitTermination}* todothat. */ public voidshutdown(){//关闭线程池 final ReentrantLock mainLock=this.mainLock;mainLock.lock();try{//如下多个操作（复合操作需要 手动方式枷锁） advanceRunState(SHUTDOWN);interruptIdleWorkers();onShutdown();// hookforScheduledThreadPoolExecutor 钩子函数}finally{mainLock.unlock();}tryTerminate();}