网站建设优化服务效果,电脑培训课程,提升学历机构,wordpress管理历史版本ThreadPoolExecutor1.什么是线程池#xff1f; #xff08;首先要理解什么是线程#xff09; 线程池#xff0c;thread pool#xff0c;是一种线程使用模式#xff0c;线程池维护着多个线程#xff0c;等待着监督管理者分配可并发执行的任务。 通俗来说#xff0c;就是…ThreadPoolExecutor1.什么是线程池 首先要理解什么是线程 线程池thread pool是一种线程使用模式线程池维护着多个线程等待着监督管理者分配可并发执行的任务。 通俗来说就是可管理和维护以及分配线程的“池子”。2.为什么使用线程池 为了减少创建和销毁线程的次数让每个线程都可以多次的使用可以根据系统情况调整线程的数量防止消耗过多内存。在实际使用中服务器在创建和销毁线程上花费的时间和消耗的系统资源都相当大使用线程池就可以优化。 线程池主要用来解决线程生命周期开销问题和资源不足问题。通过对多个任务重复使用线程线程创建的开销就被分摊到了多个任务上了而且由于在请求到达时线程已经存在所以消除了线程创建所带来的延迟。这样就可以立即为请求服务使用应用程序响应更快。另外通过适当的调整线程中的线程数目可以防止出现资源不足的情况。 通俗来说就是为了优化线程的内存开销。3.线程池的核心参数
public ThreadPoolExecutor(int corePoolSize,//核心线程数int maximumPoolSize,//最大线程数long keepAliveTime,//线程空闲时间TimeUnit unit,//时间单位BlockingQueueRunnable workQueue,//任务队列ThreadFactory threadFactory,//线程工厂RejectedExecutionHandler handler//拒绝策略)
{...
}4.线程池的执行顺序 线程池按以下行为执行任务 当线程数小于核心线程数时创建线程。 当线程数大于等于核心线程数且任务队列未满时将任务放入任务队列。 当线程数大于等于核心线程数且任务队列已满若线程数小于最大线程数创建线程。 若线程数等于最大线程数则执行拒绝策略5.线程池的参数详解
a)、corePoolSize 核心线程数默认为1。 设置规则 CPU密集型(CPU密集型也叫计算密集型指的是运算较多cpu占用高读/写I/O(硬盘/内存)较少)corePoolSize CPU核数 1 IO密集型与cpu密集型相反系统运作大部分的状况是CPU在等I/O (硬盘/内存) 的读/写操作此时CPU Loading并不高。corePoolSize CPU核数 * 2b)、maximumPoolSize 最大线程数默认为Integer.MAX_VALUE 一般设置为和核心线程数一样c)、keepAliveTime 线程空闲时间默认为60s一般设置为默认60sd)、unit 时间单位默认为秒e)、workQueue 队列当线程数目超过核心线程数时用于保存任务的队列。BlockingQueue workQueue此队列仅保存实现Runnable接口的任务。因为线程池的底层BlockingQueue的泛型为Runnable 1无界队列 队列大小无限制常用的为无界的LinkedBlockingQueue使用该队列作为阻塞队列时要尤其当心当任务耗时较长时可能会导致大量新任务在队列中堆积最终导致OOM。阅读代码发现Executors.newFixedThreadPool 采用就是 LinkedBlockingQueue而博主踩到的就是这个坑当QPS很高发送数据很大大量的任务被添加到这个无界LinkedBlockingQueue 中导致cpu和内存飙升服务器挂掉。 当然这种队列maximumPoolSize 的值也就无效了。当每个任务完全独立于其他任务即任务执行互不影响时适合于使用无界队列例如在 Web 页服务器中。这种排队可用于处理瞬态突发请求当命令以超过队列所能处理的平均数连续到达时此策略允许无界线程具有增长的可能性。 2有界队列 当使用有限的 maximumPoolSizes 时有界队列有助于防止资源耗尽但是可能较难调整和控制。常用的有两类一类是遵循FIFO原则的队列如ArrayBlockingQueue另一类是优先级队列如PriorityBlockingQueue。PriorityBlockingQueue中的优先级由任务的Comparator决定。 使用有界队列时队列大小需和线程池大小互相配合线程池较小有界队列较大时可减少内存消耗降低cpu使用率和上下文切换但是可能会限制系统吞吐量。 3同步移交队列 如果不希望任务在队列中等待而是希望将任务直接移交给工作线程可使用SynchronousQueue作为等待队列。SynchronousQueue不是一个真正的队列而是一种线程之间移交的机制。要将一个元素放入SynchronousQueue中必须有另一个线程正在等待接收这个元素。只有在使用无界线程池或者有饱和策略时才建议使用该队列。
f)、threadFactory 线程工厂用来创建线程。 为了统一在创建线程时设置一些参数如是否守护线程线程一些特性等如优先级。通过这个TreadFactory创建出来的线程能保证有相同的特性。 它是一个接口类而且方法只有一个就是创建一个线程。 如果没有另外说明则在同一个ThreadGroup 中一律使用Executors.defaultThreadFactory() 创建线程并且这些线程具有相同的NORM_PRIORITY 优先级和非守护进程状态。 通过提供不同的 ThreadFactory可以改变线程的名称、线程组、优先级、守护进程状态等等。 如果从newThread 返回 null 时ThreadFactory 未能创建线程则执行程序将继续运行但不能执行任何任务。
g)、handler 拒绝策略默认是AbortPolicy会抛出异常。 当线程数已经达到maxPoolSize且队列已满会拒绝新任务。 当线程池被调用shutdown()后会等待线程池里的任务执行完毕再shutdown。如果在调用shutdown()和线程池真正shutdown之间提交任务会拒绝新任务。 AbortPolicy 丢弃任务抛运行时异常。 CallerRunsPolicy 由当前调用的任务线程执行任务。 DiscardPolicy 忽视什么都不会发生。 DiscardOldestPolicy 从队列中踢出最先进入队列最后一个执行的任务。
6.ThreadPoolExecutor和spring封装的ThreadPoolTaskExecutor案例 ThreadPoolExecutor是Java的线程池 ThreadPoolTaskExecutor是spring封装的线程池
package com.thgy.bc.common.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.*;Slf4j
Configuration
public class ThreadPoolConfig {Beanpublic ThreadPoolTaskExecutor threadPoolTaskExecutor() {ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor();int i Runtime.getRuntime().availableProcessors();//核心线程数目executor.setCorePoolSize(i * 2);//指定最大线程数executor.setMaxPoolSize(i * 2);//队列中最大的数目executor.setQueueCapacity(i * 2 * 10);//线程名称前缀executor.setThreadNamePrefix(ThreadPoolTaskExecutor-);//rejection-policy当pool已经达到max size的时候如何处理新任务//CALLER_RUNS不在新线程中执行任务而是由调用者所在的线程来执行//对拒绝task的处理策略executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());//当调度器shutdown被调用时等待当前被调度的任务完成executor.setWaitForTasksToCompleteOnShutdown(true);//线程空闲后的最大存活时间executor.setKeepAliveSeconds(60);//加载executor.initialize();log.info(初始化线程池成功);return executor;}Beanpublic ThreadPoolExecutor threadPoolExecutor() {//获取cpu核心数int i Runtime.getRuntime().availableProcessors();//核心线程数int corePoolSize i * 2;//最大线程数int maximumPoolSize i * 2;//线程无引用存活时间long keepAliveTime 60;//时间单位TimeUnit unit TimeUnit.SECONDS;//任务队列接收一个整型的参数这个整型参数指的是队列的长度//ArrayBlockingQueue(int,boolean)boolean类型的参数是作为可重入锁的参数进行初始化默认false另外初始化了notEmpty、notFull两个信号量。BlockingQueueRunnable workQueue new ArrayBlockingQueue(i * 2 * 10);//1. 同步阻塞队列 (put,take)直接提交。直接提交策略表示线程池不对任务进行缓存。新进任务直接提交给线程池当线程池中没有空闲线程时创建一个新的线程处理此任务。// 这种策略需要线程池具有无限增长的可能性。实现为SynchronousQueue//2. 有界队列。当线程池中线程达到corePoolSize时新进任务被放在队列里排队等待处理。有界队列如ArrayBlockingQueue有助于防止资源耗尽// 但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销// 但是可能导致人工降低吞吐量。如果任务频繁阻塞例如如果它们是 I/O 边界则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小// CPU 使用率较高但是可能遇到不可接受的调度开销这样也会降低吞吐量。//3. 无界队列。使用无界队列例如不具有预定义容量的 LinkedBlockingQueue将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。// 这样创建的线程就不会超过 corePoolSize。因此maximumPoolSize 的值也就无效了。当每个任务完全独立于其他任务即任务执行互不影响时// 适合于使用无界队列例如在 Web 页服务器中。这种排队可用于处理瞬态突发请求当命令以超过队列所能处理的平均数连续到达时此策略允许无界线程具有增长的可能性。//线程工厂//defaultThreadFactory()//返回用于创建新线程的默认线程工厂。//privilegedThreadFactory()//返回一个用于创建与当前线程具有相同权限的新线程的线程工厂。ThreadFactory threadFactory Executors.defaultThreadFactory();//拒绝执行处理器RejectedExecutionHandler handler new ThreadPoolExecutor.CallerRunsPolicy();//创建线程池ThreadPoolExecutor threadPoolExecutor new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);return threadPoolExecutor;}
}
7.jdk自带的四种线程池创建方式
// 第一种线程池:固定个数的线程池,可以为每个CPU核绑定一定数量的线程数
ExecutorService fixedThreadPool Executors.newFixedThreadPool(processors * 2);
// 缓存线程池无上限
ExecutorService cachedThreadPool Executors.newCachedThreadPool();
// 单一线程池,永远会维护存在一条线程
ExecutorService singleThreadPool Executors.newSingleThreadExecutor();
// 固定个数的线程池可以执行延时任务也可以执行带有返回值的任务。
ScheduledExecutorService scheduledThreadPool Executors.newScheduledThreadPool(5);8.队列的详解 看我另一篇文章
9.调用线程池的线程案例
public ListAccountRecordVO requestTest() throws ExecutionException, InterruptedException {ListString ids Lists.newArrayList();ids.add(1);ids.add(2);ids.add(3);ids.add(4);//有返回值的情况定义接收返回值ListAccountRecordVO futureList2 Lists.newArrayList();//分布式计数器若业务不需要则可以不定义CountDownLatch countDownLatch new CountDownLatch(ids.size());for (String id : ids) {//调用线程池的线程执行任务threadPoolTaskExecutor.submit(new Runnable() {Overridepublic void run() {test(Lists.newArrayList(id),futureList2);//计数器-1countDownLatch.countDown();}});}//await阻塞直到计数器为0countDownLatch.await();System.out.println(主线程);return futureList2;}public ListAccountRecordVO test(ListString ids, ListAccountRecordVO list2){//随便写的业务逻辑代码无实际意义仅作演示System.out.println(线程体 );ListAccountRecordVO accountRecordVOS Lists.newArrayList();int i 0;AccountRecordVO accountRecordVO new AccountRecordVO();accountRecordVO.setUserId(123);accountRecordVO.setAmount(12333);for (String id : ids){accountRecordVOS.add(accountRecordVO);list2.add(accountRecordVO);}try{Thread.sleep(Long.valueOf(1000));}catch (Exception e){log.error(e.getMessage());}System.out.println(线程体结束 );return accountRecordVOS;}
