构建高性能工作线程 一 扩展 Spring 线程池
扩展 Spring 线程池
/**扩展SPRING 2.0的threadPooltaskExecutor提供返回theadPoolexuecutor的引用，用于实时调整CorePoolSize、MaximumPoolSize、KeepAliveTime三个参数。*/public class MyThreadPoolTaskExecutor implements SchedulingTaskExecutor,Executor, InitializingBean, DisposableBean { //核心线程池大小 private int corePoolSize; //最大线程池大小 private int maxPoolSize; //线程保持活动时间 private int keepAliveSeconds; //缓冲队列大小 private int queueCapacity; //线程池名称 private String threadPoolName; //thread 工厂 private ThreadFactory threadFactory; //未执行任务的接口 private RejectedExecutionHandler rejectedExecutionHandler; //jdk 线程池执行器 private ThreadPoolExecutor executorService; //用于线程同步对象 private final Object poolSizeMonitor = new Object(); //MAP 用于管理所有线程池 private static ConcurrentHashMap threadPoolMap = new ConcurrentHashMap <String,ThreadPoolExecutor>(); //默认构造函数 public MyThreadPoolTaskExecutor() { corePoolSize = 1; maxPoolSize = 2147483647; keepAliveSeconds = 60; queueCapacity = 2147483647; threadFactory = Executors.defaultThreadFactory(); rejectedExecutionHandler = new java.util.concurrent.ThreadPoolExecutor.AbortPolicy(); threadPoolName =""; } //创建缓冲队列 protected BlockingQueue createQueue(int queueCapacity) { if (this.queueCapacity > 0) return new LinkedBlockingQueue(this.queueCapacity); else return new SynchronousQueue(); } //线程池执行方法 public void execute(Runnable task) { Assert.notNull(executorService,"ThreadPoolTaskExecutor not initialized"); executorService.execute(task); } //SPRING 线程池方法扩展，返回线程池对象用于实时调整CorePoolSize、 //MaximumPoolSize、KeepAliveTime三个参数。 public ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException { Assert.state(this.executorService != null, "ThreadPoolTaskExecutor not initialized"); return this.executorService; } public void setCorePoolSize(int corePoolSize) { synchronized (this.poolSizeMonitor) { this.corePoolSize = corePoolSize; if (this.executorService != null) { this.executorService.setCorePoolSize(corePoolSize); } } } public int getCorePoolSize() { synchronized (this.poolSizeMonitor) { return this.executorService.getCorePoolSize(); } } public void setMaxPoolSize(int maxPoolSize) { synchronized (this.poolSizeMonitor) { this.maxPoolSize = maxPoolSize; if (this.executorService != null) { this.executorService.setMaximumPoolSize(maxPoolSize); } } } public int getMaxPoolSize() { synchronized (this.poolSizeMonitor) { return this.executorService.getMaximumPoolSize(); } } public void setKeepAliveSeconds(int keepAliveSeconds) { synchronized (this.poolSizeMonitor) { this.keepAliveSeconds = keepAliveSeconds; if (this.executorService != null) { this.executorService.setKeepAliveTime(keepAliveSeconds,TimeUnit.SECONDS); } } } public int getKeepAliveSeconds() { synchronized (this.poolSizeMonitor) { return this.keepAliveSeconds; } } public void setQueueCapacity(int queueCapacity) { this.queueCapacity = queueCapacity; } public void setThreadFactory(ThreadFactory threadFactory) { this.threadFactory = threadFactory == null ? Executors.defaultThreadFactory() : threadFactory; } public void setRejectedExecutionHandler( RejectedExecutionHandler rejectedExecutionHandler) { this.rejectedExecutionHandler = ((RejectedExecutionHandler)(rejectedExecutionHandler == null ? ((RejectedExecutionHandler) (new java.util.concurrent.ThreadPoolExecutor.AbortPolicy())): rejectedExecutionHandler)); }}?