线程池简单源码思路手撕实现和关于参数设置

线程池简单源码思路手撕实现

importjava.util.ArrayList;importjava.util.List;importjava.util.concurrent.BlockingQueue;importjava.util.concurrent.TimeUnit;publicclassmyThreadPool{privateintcorePoolSize;privateintmaxPoolSize;privateinttimeout;privateTimeUnittimeUnit;publicBlockingQueue<Runnable>blockingQueue;privateRejectHandlerejectHandle;privateList<Thread>coreList=newArrayList<>();privateList<Thread>supportList=newArrayList<>();publicmyThreadPool(intcorePoolSize,intmaxPoolSize,inttimeout,TimeUnittimeUnit,BlockingQueue<Runnable>blockingQueue,RejectHandlerejectHandle){this.corePoolSize=corePoolSize;this.maxPoolSize=maxPoolSize;this.timeout=timeout;this.timeUnit=timeUnit;this.blockingQueue=blockingQueue;this.rejectHandle=rejectHandle;}voidexecute(Runnablecommand){if(coreList.size()<corePoolSize){CoreThreadthread=newCoreThread(command);coreList.add(thread);thread.start();return;}if(blockingQueue.offer(command)){return;}if(coreList.size()+supportList.size()<maxPoolSize){SupportThreadthread=newSupportThread(command);supportList.add(thread);thread.start();return;}if(!blockingQueue.offer(command)){rejectHandle.reject(command,this);}}classCoreThreadextendsThread{privatefinalRunnablefirstTask;CoreThread(RunnablefirstTask){this.firstTask=firstTask;}@Overridepublicvoidrun(){firstTask.run();while(true){try{Runnablecommand=blockingQueue.take();command.run();}catch(InterruptedExceptione){thrownewRuntimeException(e);}}}}classSupportThreadextendsThread{privatefinalRunnablefirstTask;SupportThread(RunnablefirstTask){this.firstTask=firstTask;}@Overridepublicvoidrun(){firstTask.run();while(true){try{Runnablecommand=blockingQueue.poll(timeout,timeUnit);if(command==null){break;}command.run();}catch(InterruptedExceptione){thrownewRuntimeException(e);}System.out.println(Thread.currentThread().getName()+"线程死掉了");supportList.remove(Thread.currentThread());}}}}

首先是重要的线程池参数：

核心线程数：定义不会被回收的核心线程数量

最大线程数：定义除了核心线程以外的临时线程，临时线程超过时间还没任务就会被销毁

阻塞队列：选择用什么阻塞队列，使用阻塞队列是防止在核心/辅助线程再去拿任务的时候一直while对cpu空转的损害

拒绝策略：在超过最大线程就会采取对应的拒绝策略-比如报错舍弃/静默丢弃/舍弃旧任务（或者也可以考虑持久化到mysql保存起来）

时间：这个是辅助线程没任务的存活时间。

时间单位：这个是辅助线程没任务的存活时间的单位。

线程工厂：用来决定线程的命名，创建线程的方式，管理线程优先级之类的。这里只是简单demo所以没有实现。

voidexecute(Runnablecommand){if(coreList.size()<corePoolSize){CoreThreadthread=newCoreThread(command);coreList.add(thread);thread.start();return;}if(blockingQueue.offer(command)){return;}if(coreList.size()+supportList.size()<maxPoolSize){SupportThreadthread=newSupportThread(command);supportList.add(thread);thread.start();return;}if(!blockingQueue.offer(command)){rejectHandle.reject(command,this);}}

这个是重要的线程池接受任务的过程：

核心线程没满就用核心线程，满了就先放到阻塞队列，阻塞队列满了就创建临时辅助线程，辅助线程也满了就考虑拒绝策略。

classCoreThreadextendsThread{privatefinalRunnablefirstTask;CoreThread(RunnablefirstTask){this.firstTask=firstTask;}@Overridepublicvoidrun(){firstTask.run();while(true){try{Runnablecommand=blockingQueue.take();command.run();}catch(InterruptedExceptione){thrownewRuntimeException(e);}}}}classSupportThreadextendsThread{privatefinalRunnablefirstTask;SupportThread(RunnablefirstTask){this.firstTask=firstTask;}@Overridepublicvoidrun(){firstTask.run();while(true){try{Runnablecommand=blockingQueue.poll(timeout,timeUnit);if(command==null){break;}command.run();}catch(InterruptedExceptione){thrownewRuntimeException(e);}System.out.println(Thread.currentThread().getName()+"线程死掉了");supportList.remove(Thread.currentThread());}}}}

这里有firsttask的任务主要是在线程刚创建就来执行，减少再去阻塞队列拿的麻烦，核心线程和辅助线程从阻塞队列拿任务的方法分别是take()和poll()主要是后者限制时间，辅助线程执行完就会被销毁这里是发生在run()执行完。而take()永久等待出不来就不会被销毁。

然后看一下拒绝策略

packagetech.insight;/** * @author gongxuanzhangmelt@gmail.com **/publicinterfaceRejectHandle{voidreject(RunnablerejectCommand,MyThreadPoolthreadPool);}

packagetech.insight;/** * @author gongxuanzhangmelt@gmail.com **/publicclassThrowRejectHandleimplementsRejectHandle{@Overridepublicvoidreject(RunnablerejectCommand,MyThreadPoolthreadPool){thrownewRuntimeException("阻塞队列满了！");}}

packagetech.insight;/** * @author gongxuanzhangmelt@gmail.com **/publicclassDiscardRejectHandleimplementsRejectHandle{@Overridepublicvoidreject(RunnablerejectCommand,MyThreadPoolthreadPool){threadPool.blockingQueue.poll();threadPool.execute(rejectCommand);}}

只实现了报错丢弃和丢弃旧任务加入新任务。

importjava.util.concurrent.ArrayBlockingQueue;importjava.util.concurrent.TimeUnit;publicclassMain{publicstaticvoidmain(String[]args){myThreadPool myThreadPool=newmyThreadPool(2,10,1,TimeUnit.SECONDS,newArrayBlockingQueue<>(2),newThrowRejectHandle());for(inti=0;i<10;i++){finalintfi=i;myThreadPool.execute(()->{try{Thread.sleep(1000);}catch(InterruptedExceptione){thrownewRuntimeException(e);}System.out.println(Thread.currentThread().getName()+" "+fi);});}}}

main里初始化线程池，以及添加10个任务，以上就是线程池的简单实现。

接下来我记录一些自己看的扩展

关于参数该怎么设置？比如核心线程数和最大线程数和其他的？

线程数

首先看八股怎么说：

cpu密集型：核心线程数=cpu核心数加减1，最大线程数=cpu核心数加减1

io密集型：核心线程数=cpu核心数*2，最大线程数=CPU核心数 × 4 或更高

然后还有具体的场景：

比如在电商场景：流量波动极大。平时可能没什么人，一旦活动开启，流量瞬间暴涨。可以极端的把核心线程数设置为0（还是留一点但是占比小），全是临时线程没有任务就回收。

还有记录日志场景：这种一般没什么变化，就全部设置为核心线程数就好，最大线程数等于核心线程数。

阻塞队列

阻塞队列选有界的如ArrayBlockingQueue或LinkedBlockingQueue(n)，因为无界队列存在内存溢出风险，还有**SynchronousQueue：** 之前提到的电商核心线程为 0 时常用。它不存任务，直接“手递手”交给线程。

关于存活时间

电商/高并发场景：建议设置得稍长一些（如 60s）。因为流量往往是波动性的，频繁地销毁和重新创建线程是非常消耗 CPU 的。

后台低频任务：建议设置得短一些（如 1s 或 10s）。任务处理完赶紧释放资源。

默认参考：Java 自带的CachedThreadPool默认是60s，这是一个非常合理的平衡点点。

可以理解为核心线程少就长一点，多就短一点。

拒绝策略

当线程池满了（最大线程已开，队列也塞满了），新来的任务怎么办？Java 提供了四种标准策略：

AbortPolicy（默认）：直接抛出RejectedExecutionException。
- 适用：关键业务，必须让调用方知道任务失败了。
CallerRunsPolicy（调用者运行）：谁提交的任务，谁自己去执行。
- 适用：最重要的兜底方案。它会减慢生产者的速度（因为生产者去干活了，没空发任务了），起到一种天然的“降级”和“限流”作用。
DiscardPolicy：直接丢掉任务，不报任何错。
- 适用：比如打印无关紧要的日志、监控埋点，丢了就丢了。
DiscardOldestPolicy：丢弃队列里排队最久（最老）的任务，把当前任务塞进去。
- 适用：具有“时效性”的任务，比如实时行情数据，旧的数据没意义了。