重庆分公司,新征程启航
为企业提供网站建设、域名注册、服务器等服务
ThreadPoolExecutor
创新互联建站长期为千余家客户提供的网站建设服务,团队从业经验10年,关注不同地域、不同群体,并针对不同对象提供差异化的产品和服务;打造开放共赢平台,与合作伙伴共同营造健康的互联网生态环境。为东昌企业提供专业的做网站、网站设计,东昌网站改版等技术服务。拥有十载丰富建站经验和众多成功案例,为您定制开发。
ThreadPoolExecutor线程池,java提供开发框架,管理线程的创建、销毁、优化、监控等。
有4种不同的任务队列:
1.ArrayBlockingQueue:基于数组结构的任务队列。此队列按先进先出的原则对任务进行排序。
2.LinkedBlockingQueue:基于链表结构的任务队列。此队列也是按先进先出的原则对任务进行排序。但性能比ArrayBlockingQueue高。
3.synchronousQueue:不存储元素的任务队列。每个插入操作必须等到另一个线程调用移除操作,否则插入操作一直处于阻塞状态。
4.PriorityBlockingQueue:具有优先级的任务队列。此队列中的元素必须能够比较。
拒绝策略:
RejectedExecutionHandler(饱和策略 ):当线程池中的线程数大于maximumPoolSize时,线程池就不能在处理任何任务了,这时线程池会抛出异常。原因就是这个策略默认情况下是AbortPolicy:表示无法处理新任务时抛出异常。
1.AbortPolicy:直接抛出异常。
2.CallerRunsPolicy:只用调用者所在线程来运行任务。
3.DiscardOldestPolicy:丢弃队列里最近的一个任务,并执行当前任务
4.DiscardPolicy:不处理,丢弃掉。
自定义:
ThreadPoolExecutor.AbortPolicy()
//抛出java.util.concurrent.RejectedExecutionException异常
ThreadPoolExecutor.CallerRunsPolicy()
//重试添加当前的任务,他会自动重复调用execute()方法
ThreadPoolExecutor.DiscardOldestPolicy()
//抛弃旧的任务
ThreadPoolExecutor.DiscardPolicy()
// 抛弃当前的任务
private static class RecjectThreadHandler implements RejectedExecutionHandler { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { } // 异常记录 private void doLog(Runnable r, ThreadPoolExecutor executor) { System.out.println(r.toString()+"excutor failed."+executor.getCompletedTaskCount()); } }
创建线程工厂:
用来创建线程。
public class CheckThreadFactory implements ThreadFactory { private String threadGroupName; private AtomicInteger count = new AtomicInteger(0); public CheckThreadFactory(String threadGroupName) { this.threadGroupName = threadGroupName; } @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName(threadGroupName+"--"+count.addAndGet(1)); thread.setPriority(5); thread.setDaemon(true);.// 设置为守护线程, 默认为主线程 return thread; } }
线程工具类:
/** * @author Donald * @create 2019-09-21 21:40 */ public class CheckExcetPool { // 线程池核心线程数 private static int corePoolSize = Runtime.getRuntime().availableProcessors() * 5; // 最大线程数 private static int maximumPoolSize = corePoolSize > 255 ? 255 : corePoolSize * 2; // 线程池中除了核心线程,其他线程的最大存活时间 private static int keepAliveTime = 60; // 时间单位 private static TimeUnit timeUnit = TimeUnit.SECONDS; // 线程等待队列 private static BlockingQueue queue = new LinkedBlockingQueue(); //private static BlockingQueue queue = new ArrayBlockingQueue(30); // 创建线程的工厂 private static CheckThreadFactory checkThreadFactory = new CheckThreadFactory("checkGroup"); // 拒绝策略 当提交任务数超过maxmumPoolSize+workQueue之和时, // * 即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)), // * 任务会交给RejectedExecutionHandler来处理 /*handler的拒绝策略: 有四种:第一种AbortPolicy:不执行新任务,直接抛出异常,提示线程池已满 第二种DisCardPolicy:不执行新任务,也不抛出异常 第三种DisCardOldSetPolicy:将消息队列中的第一个任务替换为当前新进来的任务执行 第四种CallerRunsPolicy:直接调用execute来执行当前任务*/ private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, queue, checkThreadFactory ); public static void submit( Runnable runnable) { System.out.println(corePoolSize+"::"+queue.size()); threadPoolExecutor.submit(runnable); } public static Future submit(Callable callable) { return threadPoolExecutor.submit(callable); } public static void excutor( Runnable run, T result ) { threadPoolExecutor.submit( run,result ); } public static void excutor( Runnable run) { threadPoolExecutor.execute( run); } private static class RecjectThreadHandler implements RejectedExecutionHandler { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { } // 异常记录 private void doLog(Runnable r, ThreadPoolExecutor executor) { System.out.println(r.toString()+"excutor failed."+executor.getCompletedTaskCount()); } } }
线程服务类,实现runnable 接口:
/** * @author Donald * @create 2019-09-21 23:00 */ public class ThreadService implements Runnable { private CountDownLatch countDownLatch; private UserInterface userInterface; public ThreadService(CountDownLatch countDownLatch, UserInterface userInterface) { this.countDownLatch = countDownLatch; this.userInterface = userInterface; } @Override public void run() { try { long start = System.currentTimeMillis(); userInterface.doSomething(); System.err.println(String.format("user time :%s",System.currentTimeMillis()-start)); Thread.sleep(1000); }catch ( Exception e) { e.printStackTrace(); }finally { countDownLatch.countDown(); } } }
具体业务逻辑:
/** * @author Donald * @create 2019-09-21 22:51 */ public interface UserInterface { void doSomething(); }
业务类:
/** * @author Donald * @create 2019-09-21 22:51 */ public class UserService implements UserInterface { private int number; public UserService(int number) { this.number = number; } @Override public void doSomething() { System.out.println(Thread.currentThread().getName()+"<<<<"+number); } }
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持创新互联。