Spring任务执行-TaskExecutor

概述

Spring框架提供了任务执行器和任务调度器的接口：TaskExecutor和 TaskScheduler 来支持任务异步执行和任务调度的功能。同时使用框架自己定义的抽象接口来屏蔽掉底层JDK版本间以及Java EE中的线程池和定时任务处理的差异。

另外Spring还支持集成JDK内部的定时器Timer和第三方提供的定时器Quartz Scheduler框架。

任务执行接口：TaskExecutor

public interface TaskExecutor extends Executor

最初创建TaskExecutor是为了在需要时为其他Spring组件提供线程池抽象。在Spring的一些其他组件中比如ApplicationEventMulticaster，Quartz都是使用TaskExecutor来作为线程池的抽象的。但是，如果您的bean需要线程池行为，则可以根据⾃⼰的需要使⽤此抽象。

常用四个实现类：SimpleAsyncTaskExecutor、ConcurrentTaskExecutor、ThreadPoolTaskExecutor、SimpleThreadPoolTaskExecutor 等。

SimpleAsyncTaskExecutor

此实现支持任务的异步执行，但是此实现没有线程的复用，每次执行一个提交的任务时候都会新建一个线程，任务执行完成后会将线程关闭，最大并发数默认是没有限制的，但是可以通过方法 setConcurrencyLimit(int concurrencyLimit)来设置最大并发数。

一般使用 线程池ThreadPoolTaskExecutor 来代替此实现，特别是执行一些生命周期很短的任务的时候。

ConcurrentTaskExecutor

这个实现是java.util.concurrent.Executor的适配器实例。一般使用 线程池ThreadPoolTaskExecutor 来代替，它将Executor配置参数公开为bean属性。

很少需要直接使⽤ConcurrentTaskExecutor，但是如果ThreadPoolTaskExecutor不够灵活，不能满⾜您的需求，那么ConcurrentTaskExecutor是另一种选择。

ThreadPoolTaskExecutor

这个实现是最常见的。它公开了用于配置java.util.concurrent的bean属性，ThreadPoolExecutor将其包装在一个TaskExecutor中，此实现可以通过属性注入来配置线程池的相关配置。

SimpleThreadPoolTaskExecutor

当我们有Quartz和非Quartz共享同一个线程池的需求的时候使用SimpleThreadPoolTaskExecutor。

xml配置Demo

定义一个任务DataSimulation：

package TaskExecutorDemo;
import org.springframework.stereotype.Component;
import java.util.Random;

@Component
public class DataSimulation implements Runnable {
    @Override
    public void run() {
        Random random = new Random();
        System.out.println("[" + Thread.currentThread().getName() + "]" +"-" + random.nextInt(10));
    }
}

定一个类，注入TaskExecutor：

package TaskExecutorDemo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

@Component
public class DataFacotory {
    @Autowired
    public TaskExecutor executor;

    public TaskExecutor getExecutor() {
        return executor;
    }

    public void setExecutor(TaskExecutor executor) {
        this.executor = executor;
    }

    public void dataFactory(){

        for (int i =0; i < 10; i++){
            executor.execute(new DataSimulation());
        }
    }
}

applicationContext.xml中定义bean id = “taskExecutor”（id名称随便设置），并配置ThreadPoolTaskExecutor的属性：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="
         http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
         http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">

    <context:component-scan base-package="TaskExecutorDemo"/>
    <bean id = "taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
        <property name="corePoolSize" value = "5"></property>
        <property name = "maxPoolSize" value="10"></property>
        <property name="queueCapacity" value="25"></property>
    </bean>
</beans>

调用：

public class testTaskExecutor {
    public static void main(String[] args) {
        ApplicationContext context = new ClassPathXmlApplicationContext(new String[] {"TaskExecutorDemo/applicationContext.xml"});

        DataFacotory df = (DataFacotory)context.getBean("dataFacotory");
        df.dataFactory();
    }
}

运行结果：

完成配置后即可使用此线程池。

@Autowired public TaskExecutor executor 为什么可以注入配置的线程池 ？

首先@Autowired 根据类型注入的。
public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport implements SchedulingTaskExecutor
public interface SchedulingTaskExecutor extends AsyncTaskExecutor
public interface AsyncTaskExecutor extends TaskExecutor
public interface TaskExecutor extends Executor
所以，类ThreadPoolTaskExecutor继承TaskExecutor，最终继承Executor。@Autowired 根据类型注入可以使用父类声明。
当然还可以使用最终接口Executor声明变量。和bean配置id = "taskExecutor"没有任何关系。

@Component
public class DataFacotory {
    @Autowired
    public Executor executor;

    public Executor getExecutor() {
        return executor;
    }

    public void setExecutor(Executor executor) {
        this.executor = executor;
    }

    public void dataFactory(){

        for (int i =0; i < 10; i++){
            executor.execute(new DataSimulation());
        }
    }
}

最终结果一样。

Spring提供的线程池可以通过配置文件配置线程池的配置，相比JDK自带的线程池是一个很大的优势。

配置类Demo

新增配置类ExecuteConfiguration：

@Configuration
public class ExecuteConfiguration {
    @Bean
    public Executor getAsyncExecutor() {
        //使用Spring内置线程池任务对象
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //设置线程池参数
        taskExecutor.setCorePoolSize(5);//核心池大小
        taskExecutor.setMaxPoolSize(10);//最大线程数
        taskExecutor.setQueueCapacity(25);//队列程度
        taskExecutor.setKeepAliveSeconds(100);//空闲时间
        taskExecutor.setThreadNamePrefix("Thread_Test");//线程前缀名称
        taskExecutor.initialize();
        return taskExecutor;
    }
}

applicationContext.xml删除原定义的bean

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="
         http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
         http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">

    <context:component-scan base-package="TaskExecutorDemo"/>

</beans>

运行结果：

好处

• 通过使用线程池来实现线程的复用，减少线程创建和销毁的开销。
• 将执行线程的任务交给线程池来操作，一定意义上实现了解耦。
• 使用线程池可以控制任务的最大并发数目，这个在防止内存溢出以及并发优化方面有很重要的作用。