SSM配置和使用ThreadPoolTaskExecutor线程池

为什么使用线程池？
提升性能：它们通常在执行大量异步任务时，由于减少了每个任务的调用开销，并且它们提供了一种限制和管理资源（包括线程）的方法，使得性能提升明显

在配置使用ThreadPoolTaskExecutor前，先了解ThreadPoolExecutor，以及它们两个的区别

一、ThreadPoolExecutor

1、 ThreadPoolExecutor是jdk线程池中的一个类，它继承了AbstractExecutorService类，而AbstractExecutorService类实现了Executorservice接口，Executorservice继承了Executor接口

2、Executor接口
Executor接口是线程池最顶层的一个接口，它只有一个方法void execute(Runnable command)： 在未来某个时间执行给定的命令。该命令可能在新的线程、已入池的线程或者正调用的线程中执行，这由 Executor 实现决定。

3、ExecutorService接口
ExecutorService是一个接口，提供了管理终止的方法，以及可为跟踪一个或多个异步任务执行状况而生成Future 的方法。
它有三个实现类：AbstractExecutorService（默认实现类） , ScheduledThreadPoolExecutor和 ThreadPoolExecutor。
Executors 提供了此接口的几种常用实现的工厂方法，用来创建合适的线程池，返回ExecutorService类型的线程池。方法如下：
ExecutorService newFixedThreadPool() : 创建固定大小的线程池
ExecutorService newCachedThreadPool() : 缓存线程池，线程池的数量不固定，可以根据需求自 动的更改数量。
ExecutorService newSingleThreadExecutor() : 创建单个线程池。 线程池中只有一个线程
ScheduledExecutorService newScheduledThreadPool() : 创建固定大小的线程，可以延迟或定时的执行。

public class Executors {

/**
* Creates a thread pool that reuses a fixed number of threads
* operating off a shared unbounded queue.  At any point, at most
* {@code nThreads} threads will be active processing tasks.
* If additional tasks are submitted when all threads are active,
* they will wait in the queue until a thread is available.
* If any thread terminates due to a failure during execution
* prior to shutdown, a new one will take its place if needed to
* execute subsequent tasks.  The threads in the pool will exist
* until it is explicitly {@link ExecutorService#shutdown shutdown}.
*
* @param nThreads the number of threads in the pool
* @return the newly created thread pool
* @throws IllegalArgumentException if {@code nThreads <= 0}
*/
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}

/**
* Creates a thread pool that maintains enough threads to support
* the given parallelism level, and may use multiple queues to
* reduce contention. The parallelism level corresponds to the
* maximum number of threads actively engaged in, or available to
* engage in, task processing. The actual number of threads may
* grow and shrink dynamically. A work-stealing pool makes no
* guarantees about the order in which submitted tasks are
* executed.
*
* @param parallelism the targeted parallelism level
* @return the newly created thread pool
* @throws IllegalArgumentException if {@code parallelism <= 0}
* @since 1.8
*/
public static ExecutorService newWorkStealingPool(int parallelism) {
return new ForkJoinPool
(parallelism,
ForkJoinPool.defaultForkJoinWorkerThreadFactory,
null, true);
}

一、ThreadPoolTaskExecutor

类名org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
从类名可以看出这是spring提供的线程池

从它的初始化方法中可以看出，它的底层是ThreadPoolExecutor

@Override
protected ExecutorService initializeExecutor(
ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
ThreadPoolExecutor executor  = new ThreadPoolExecutor(
this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
queue, threadFactory, rejectedExecutionHandler);
if (this.allowCoreThreadTimeOut) {
executor.allowCoreThreadTimeOut(true);
}

this.threadPoolExecutor = executor;
return executor;
}

1.ThreadPoolTaskExecutor的简介

ThreadPoolTaskExecutor是InitializingBean、DisposableBean的实现类，
spring容器后会自动处理其初始化方法和注销方法，我们只需配置bean即可。

线程池的单例问题，交给spring容器实现单例，不用自己实现单例

2.使用基于xml的方式创建配置

<!-- spring线程池 -->
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<!-- 核心线程数  -->
<property name="corePoolSize" value="5"/>
<!-- 最大线程数 -->
<property name="maxPoolSize" value="10"/>
<!-- 队列最大长度 >=mainExecutor.maxSize -->
<property name="queueCapacity" value="999999"/>
<!-- 线程池维护线程所允许的空闲时间 -->
<property name="keepAliveSeconds" value="20"/>
<!-- 线程名字前缀 -->
<property name="threadNamePrefix" value="test_executor"/>
<!-- 线程池对拒绝任务(无线程可用)的处理策略 -->
<property name="rejectedExecutionHandler">
<bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy"/>
</property>
</bean>

ThreadPoolTaskExecutor的拒绝策略：

rejectedExectutionHandler参数字段用于配置绝策略，常用拒绝策略如下

AbortPolicy：用于被拒绝任务的处理程序，它将抛出RejectedExecutionException

CallerRunsPolicy：用于被拒绝任务的处理程序，它直接在execute方法的调用线程中运行被拒绝的任务。

DiscardOldestPolicy：用于被拒绝任务的处理程序，它放弃最旧的未处理请求，然后重试execute。

DiscardPolicy：用于被拒绝任务的处理程序，默认情况下它将丢弃被拒绝的任务。

提交任务：

无返回值的任务使用execute(Runnable)
有返回值的任务使用submit(Runnable)

无返回值测试：

@Service("demo")
public class Demo {

@Autowired
private ThreadPoolTaskExecutor taskExecutor;

public void async(){
taskExecutor.execute(new Runnable() {
@Override
public void run() {
System.out.println("线程" + Thread.currentThread().getName() + "执行异步线程" );
}
});
}
}

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration({"classpath:spring-context.xml"})
@ActiveProfiles("development")
public class TestTest {

@Autowired
private Demo demo;

@Test
public void test(){
System.out.println("开始");
for (int i = 1; i <= 10; i++) {
demo.async();
}
System.out.println("结束");
}

}

结果：

执行流程：

1.当一个任务被提交到线程池时，首先查看线程池的核心线程是否都在执行任务，否就选择一条线程执行任务，是就执行第二步。
2.查看核心线程池是否已满，不满就创建一条线程执行任务，否则执行第三步。
3.查看任务队列是否已满，不满就将任务存储在任务队列中，否则执行第四步。
4.查看线程池是否已满，不满就创建一条线程执行任务，否则就按照策略处理无法执行的任务。
在ThreadPoolExecutor中表现为:
1.如果当前运行的线程数小于corePoolSize，那么就创建线程来执行任务（执行时需要获取全局锁）。
2.如果运行的线程大于或等于corePoolSize，那么就把task加入BlockQueue。
3.如果创建的线程数量大于BlockQueue的最大容量，那么创建新线程来执行该任务。
4.如果创建线程导致当前运行的线程数超过maximumPoolSize，就根据饱和策略来拒绝该任务。

springboot中@Async多线程注解使用

• 点赞 2
• 收藏
• 分享
• 文章举报