java线程池教程,java线程池调优

2022-07-31 09:26:12 分类：java 阅读()

　　java线程池教程,java线程池调优

　　

目录

升级版线程池的优化线程池构造器线程池拒绝策略执行方法手写线程池源码myexecutorservicemyrejectedexecutionexception myrejectedexecutionhandle核心类MyThreadPoolExecutor线程池测试类

 

　　

升级版线程池的优化

1:新增了4种拒绝策略。分别为：MyAbortPolicy、MyDiscardPolicy、MyDiscardOldestPolicy、MyCallerRunsPolicy

 

　　2:对线程池MyThreadPoolExecutor的构造方法进行优化，增加了参数校验，防止乱传参数现象。

　　3:这是最重要的一个优化。

　　移除线程池的线程预热功能。因为线程预热会极大的耗费内存，当我们不用线程池时也会一直在运行状态。换来的是在调用执行方法添加任务时通过检查工人线程集合目前的大小与最小雇佣数量的值去比较，再通过新的我的工人()去创建添加线程到线程池，这样好处就是当我们创建线程池如果不使用的话则对当前内存没有一点影响，当使用了才会创建线程并放入线程池中进行复用。

　　

线程池构造器

public MyThreadPoolExecutor(){ this(5，new ArrayBlockingQueue(10)，Executors.defaultThreadFactory()，默认句柄)；} public MyThreadPoolExecutor(int corePoolSize，BlockingQueueRunnable waitingQueue，ThreadFactory ThreadFactory){ this(corePoolSize，waitingQueue，ThreadFactory，默认句柄)；} public MyThreadPoolExecutor(int corePoolSize，BlockingQueueRunnable等待队列，ThreadFactory threadFactory，MyRejectedExecutionHandle handle){ this。workers=new HashSet(corePoolSize)；if(corePoolSize=0waitingQueue！=nullthreadFactory！=nullhandle！=null){这个。corePoolSize=corePoolSize这个。等待队列=等待队列；这个。螺纹工厂=螺纹工厂；this . handle=handle } else { throw new NullPointerException(线程池参数不合法);} }

 

　　

线程池拒绝策略

策略接口：MyRejectedExecutionHandle

 

　　包com。spring框架。并发；/** * 自定义拒绝策略* @作者游政杰*/公共接口MyRejectedExecutionHandle { void拒绝执行(Runnable Runnable，MyThreadPoolExecutor threadPoolExecutor)；}策略内部实现类

　　/** * 实现自定义拒绝策略*///抛异常策略(默认)公共静态类我的生育政策实现MyRejectedExecutionHandle { public MyAbortPolicy(){ } @ Override public void rejected execution(Runnable r，MyThreadPoolExecutor t){ throw new MyRejectedExecutionException(任务- r.toString()被线程池- t.toString()拒绝);} } //默默

　　丢弃策略 public static class MyDiscardPolicy implements MyRejectedExecutionHandle{ public MyDiscardPolicy() { } @Override public void rejectedExecution(Runnable runnable, MyThreadPoolExecutor threadPoolExecutor) { } } //丢弃掉最老的任务策略 public static class MyDiscardOldestPolicy implements MyRejectedExecutionHandle{ public MyDiscardOldestPolicy() { } @Override public void rejectedExecution(Runnable runnable, MyThreadPoolExecutor threadPoolExecutor) { if(!threadPoolExecutor.isShutdown()){ //如果线程池没被关闭 threadPoolExecutor.getWaitingQueue().poll();//丢掉最老的任务，此时就有位置当新任务了 threadPoolExecutor.execute(runnable); //把新任务加入到队列中 } } } //由调用者调用策略 public static class MyCallerRunsPolicy implements MyRejectedExecutionHandle{ public MyCallerRunsPolicy(){ } @Override public void rejectedExecution(Runnable runnable, MyThreadPoolExecutor threadPoolExecutor) { if(!threadPoolExecutor.isShutdown()){//判断线程池是否被关闭 runnable.run(); } } }封装拒绝方法

　　

 protected final void reject(Runnable runnable){ this.handle.rejectedExecution(runnable, this); } protected final void reject(Runnable runnable,MyThreadPoolExecutor threadPoolExecutor){ this.handle.rejectedExecution(runnable, threadPoolExecutor); }

 

　　

execute方法

 @Override public boolean execute(Runnable runnable) { if (!this.waitingQueue.offer(runnable)) { this.reject(runnable); return false; } else { if(this.workers!=null&&this.workers.size()<corePoolSize){//这种情况才能添加线程 MyWorker worker = new MyWorker(); //通过构造方法添加线程 } return true; } }

可以看出只有当往线程池放任务时才会创建线程对象。

 

　　

 

　　

手写线程池源码

 

　　

MyExecutorService

package com.springframework.concurrent;import java.util.concurrent.BlockingQueue;/** * 自定义线程池业务接口 * @author 游政杰 */public interface MyExecutorService { boolean execute(Runnable runnable); void shutdown(); void shutdownNow(); boolean isShutdown(); BlockingQueue<Runnable> getWaitingQueue();}

 

　　

MyRejectedExecutionException

package com.springframework.concurrent;/** * 自定义拒绝异常 */public class MyRejectedExecutionException extends RuntimeException { public MyRejectedExecutionException() { } public MyRejectedExecutionException(String message) { super(message); } public MyRejectedExecutionException(String message, Throwable cause) { super(message, cause); } public MyRejectedExecutionException(Throwable cause) { super(cause); }}

 

　　

MyRejectedExecutionHandle

package com.springframework.concurrent;/** * 自定义拒绝策略 * @author 游政杰 */public interface MyRejectedExecutionHandle { void rejectedExecution(Runnable runnable,MyThreadPoolExecutor threadPoolExecutor);}

 

　　

核心类MyThreadPoolExecutor

package com.springframework.concurrent;import java.util.HashSet;import java.util.Set;import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.BlockingQueue;import java.util.concurrent.Executors;import java.util.concurrent.ThreadFactory;import java.util.concurrent.atomic.AtomicInteger;/** * 纯手撸线程池框架 * @author 游政杰 */public class MyThreadPoolExecutor implements MyExecutorService{ private static final AtomicInteger taskcount=new AtomicInteger(0);//执行任务次数 private static final AtomicInteger threadNumber=new AtomicInteger(0); //线程编号 private static volatile int corePoolSize; //核心线程数 private final HashSet<MyWorker> workers; //工作线程 private final BlockingQueue<Runnable> waitingQueue; //等待队列 private static final String THREADPOOL_NAME="MyThread-Pool-";//线程名称 private volatile boolean isRunning=true; //是否运行 private volatile boolean STOPNOW=false; //是否立刻停止 private volatile ThreadFactory threadFactory; //线程工厂 private static final MyRejectedExecutionHandle defaultHandle=new MyThreadPoolExecutor.MyAbortPolicy();//默认拒绝策略 private volatile MyRejectedExecutionHandle handle; //拒绝紫略 public MyThreadPoolExecutor(){ this(5,new ArrayBlockingQueue<>(10), Executors.defaultThreadFactory(),defaultHandle); } public MyThreadPoolExecutor(int corePoolSize, BlockingQueue<Runnable> waitingQueue,ThreadFactory threadFactory) { this(corePoolSize,waitingQueue,threadFactory,defaultHandle); } public MyThreadPoolExecutor(int corePoolSize, BlockingQueue<Runnable> waitingQueue,ThreadFactory threadFactory,MyRejectedExecutionHandle handle) { this.workers=new HashSet<>(corePoolSize); if(corePoolSize>=0&&waitingQueue!=null&&threadFactory!=null&&handle!=null){ this.corePoolSize=corePoolSize; this.waitingQueue=waitingQueue; this.threadFactory=threadFactory; this.handle=handle; }else { throw new NullPointerException("线程池参数不合法"); } } /** * 实现自定义拒绝策略 */ //抛异常策略（默认） public static class MyAbortPolicy implements MyRejectedExecutionHandle{ public MyAbortPolicy(){ } @Override public void rejectedExecution(Runnable r, MyThreadPoolExecutor t) { throw new MyRejectedExecutionException("任务-> "+r.toString()+"被线程池-> "+t.toString()+" 拒绝"); } } //默默丢弃策略 public static class MyDiscardPolicy implements MyRejectedExecutionHandle{ public MyDiscardPolicy() { } @Override public void rejectedExecution(Runnable runnable, MyThreadPoolExecutor threadPoolExecutor) { } } //丢弃掉最老的任务策略 public static class MyDiscardOldestPolicy implements MyRejectedExecutionHandle{ public MyDiscardOldestPolicy() { } @Override public void rejectedExecution(Runnable runnable, MyThreadPoolExecutor threadPoolExecutor) { if(!threadPoolExecutor.isShutdown()){ //如果线程池没被关闭 threadPoolExecutor.getWaitingQueue().poll();//丢掉最老的任务，此时就有位置当新任务了 threadPoolExecutor.execute(runnable); //把新任务加入到队列中 } } } //由调用者调用策略 public static class MyCallerRunsPolicy implements MyRejectedExecutionHandle{ public MyCallerRunsPolicy(){ } @Override public void rejectedExecution(Runnable runnable, MyThreadPoolExecutor threadPoolExecutor) { if(!threadPoolExecutor.isShutdown()){//判断线程池是否被关闭 runnable.run(); } } } //call拒绝方法 protected final void reject(Runnable runnable){ this.handle.rejectedExecution(runnable, this); } protected final void reject(Runnable runnable,MyThreadPoolExecutor threadPoolExecutor){ this.handle.rejectedExecution(runnable, threadPoolExecutor); } /** * MyWorker就是我们每一个线程对象 */ private final class MyWorker implements Runnable{ final Thread thread; //为每个MyWorker MyWorker(){ Thread td = threadFactory.newThread(this); td.setName(THREADPOOL_NAME+threadNumber.getAndIncrement()); this.thread=td; this.thread.start(); workers.add(this); } //执行任务 @Override public void run() { //循环接收任务 while (true) { //循环退出条件： //1：当isRunning为false并且waitingQueue的队列大小为0（也就是无任务了），会优雅的退出。 //2：当STOPNOW为true，则说明调用了shutdownNow方法进行暴力退出。 if((!isRunning&&waitingQueue.size()==0)STOPNOW) { break; }else { //不断取任务，当任务！=null时则调用run方法处理任务 Runnable runnable = waitingQueue.poll(); if(runnable!=null){ runnable.run(); System.out.println("task==>"+taskcount.incrementAndGet()); } } } } } //往线程池中放任务 @Override public boolean execute(Runnable runnable) { if (!this.waitingQueue.offer(runnable)) { this.reject(runnable); return false; } else { if(this.workers!=null&&this.workers.size()<corePoolSize){//这种情况才能添加线程 MyWorker worker = new MyWorker(); //通过构造方法添加线程 } return true; } } //优雅的关闭 @Override public void shutdown() { this.isRunning=false; } //暴力关闭 @Override public void shutdownNow() { this.STOPNOW=true; } //判断线程池是否关闭 @Override public boolean isShutdown() { return !this.isRunningSTOPNOW; } //获取等待队列 @Override public BlockingQueue<Runnable> getWaitingQueue() { return this.waitingQueue; }}

 

　　

线程池测试类

package com.springframework.test;import com.springframework.concurrent.MyThreadPoolExecutor;import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.Executors;public class ThreadPoolTest { public static void main(String[] args) {// MyThreadPoolExecutor myThreadPoolExecutor = new MyThreadPoolExecutor// (5,new ArrayBlockingQueue<>(6), Executors.defaultThreadFactory(),new MyThreadPoolExecutor.MyAbortPolicy());// MyThreadPoolExecutor myThreadPoolExecutor = new MyThreadPoolExecutor// (5,new ArrayBlockingQueue<>(6), Executors.defaultThreadFactory(),new MyThreadPoolExecutor.MyDiscardPolicy());// MyThreadPoolExecutor myThreadPoolExecutor = new MyThreadPoolExecutor// (5,new ArrayBlockingQueue<>(6), Executors.defaultThreadFactory(),new MyThreadPoolExecutor.MyDiscardOldestPolicy()); MyThreadPoolExecutor myThreadPoolExecutor = new MyThreadPoolExecutor (5,new ArrayBlockingQueue<>(6), Executors.defaultThreadFactory(),new MyThreadPoolExecutor.MyCallerRunsPolicy()); for(int i=0;i<11;i++){ int finalI = i; myThreadPoolExecutor.execute(()->{ System.out.println(Thread.currentThread().getName()+">>>>"+ finalI); }); } myThreadPoolExecutor.shutdown();// myThreadPoolExecutor.shutdownNow(); }}

好了升级版线程池就优化到这了，后面可能还会出完善版，不断进行优化。

 

　　到此这篇关于非常适合新手学生的Java线程池升级版的文章就介绍到这了,更多相关Java 线程池内容请搜索盛行IT以前的文章或继续浏览下面的相关文章希望大家以后多多支持盛行IT！

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。

标签：

相关文章阅读

  • rxjava是如何实现线程切换的,rxjava线程切换原理面试
  • rxjava是如何实现线程切换的,rxjava线程切换原理面试,RxJava的消息发送和线程切换实现原理
  • java创建线程的三种方式,java线程里面创建线程,一文搞懂Java创建线程的五种方法
  • ,,Java线程间的通信方式详解
  • ,,Java线程休眠的5种方法
  • java线程里面创建线程,java实现多线程
  • java线程池并发执行,Java线程池实现原理
  • 线程的几种状态分别是什么,以下哪些属于java线程的状态
  • java线程同步是什么意思,java线程同步的几种方法
  • java线程的作用,线程有啥用
  • Java线程安全,什么是类的线程安全
  • java线程池的组成和作用,Java线程池的用法
  • 以下哪些属于java线程的状态,线程的几种状态分别是什么
  • java如何创建线程锁,java线程里面创建线程
  • 什么是java线程池,什么是java线程同步
    • 留言与评论（共有 条评论）
       
    验证码：