spring学习（二十八）--@Async线程池

之前使用java线程池，都是先自己通过实现ThreadPoolExecutor写线程池类，然后写多线程类，用线程池的execute方法去执行多线程类类。

在spring中，可以通过@Async(value="beanId")注解来使用线程池进行多线程编程。

线程池的创建有两种方式，一种是配置文件，一种是通过java编程。

1、通过配置文件创建线程池，在spring配置文件中加入下面配置，配置文件要开启多线程类所在包的注解扫描：

<?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:aop="http://www.springframework.org/schema/aop"
       xmlns:tx="http://www.springframework.org/schema/tx" xmlns:jdbc="http://www.springframework.org/schema/jdbc"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:mvc="http://www.springframework.org/schema/mvc" xmlns:task="http://www.springframework.org/schema/task"
       xsi:schemaLocation="http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd
     http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
     http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-3.1.xsd
     http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.1.xsd
     http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.1.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd"
       default-autowire="byName">

    <description>我的spring线程池配置</description>

    <!-- 配置@Async注解缺省参数时候的默认异步任务线程池为myExecutor -->
    <task:annotation-driven executor="myExecutor"/>

　　<!-- 第一个线程池 -->
    <task:executor id="asyncExecutor" pool-size="100-10000" queue-capacity="10"/>

    <!-- 第二个线程池 -->
    <task:executor id="myExecutor" pool-size="15-50" queue-capacity="100" keep-alive="60" rejection-policy="CALLER_RUNS"/>
</beans>

<task:executor />配置参数：

id：当配置多个executor时，被@Async("id")指定使用；也被作为线程名的前缀。

pool-size：格式"core size-max size"

core size：最小的线程数，缺省：1

max size：最大的线程数，缺省：Integer.MAX_VALUE

queue-capacity：当最小的线程数已经被占用满后，新的任务会被放进queue里面，当这个queue的capacity也被占满之后，pool里面会创建新线程处理这个任务，直到总线程数达到了max size，这时系统会拒绝这个任务并抛出TaskRejectedException异常（缺省配置的情况下，可以通过rejection-policy来决定如何处理这种情况）。缺省值为：Integer.MAX_VALUE

keep-alive：超过core size的那些线程，任务完成后，再经过这个时长（秒）会被结束掉

rejection-policy：当pool已经达到max size的时候，如何处理新任务

ABORT（缺省）：抛出TaskRejectedException异常，然后不执行DISCARD：不执行，也不抛出异常

DISCARD_OLDEST：丢弃queue中最旧的那个任务

CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行

在方法中加上@Async注解

package cn.leadeon.message.test;

import org.springframework.context.annotation.ImportResource;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

@Component
@ImportResource("classpath:/config/spring-threadpool.xml")
public class AsyncTest {

　　 //这个方法注解缺省了value参数，使用默认的myExecutor作为线程池
    @Async
    public void test1() {

        System.out.println("异步执行test1！！！");
        System.out.println("线程id：" + Thread.currentThread().getId());
        System.out.println("线程名称：" + Thread.currentThread().getName());

    }

　　//这个方法有参数(value="asyncExecutor")，使用asyncExecutor作为线程池

　　@Async(value="asyncExecutor")
    public void test2() {
        System.out.println("异步执行test2！！！");
        System.out.println("线程id：" + Thread.currentThread().getId());
        System.out.println("线程名称：" + Thread.currentThread().getName());
    }

}

当在外部调用testAsync方法时即在新线程中执行，由上面<task: annotation-driven/>执行器去维护线程。

总结：先用context:component-scan去扫描注解，让spring能识别到@Async注解，然后task:annotation-driven去驱动@Async注解，并可以指定默认的线程执行器executor。那么当用@Async注解的方法或类得到调用时，线程执行器会创建新的线程去执行。

2、注意点：@EnableAsync注解与<task:annotation-driven executor="messageExecutor"/>等价，两者只能使用其一，不然启动会报错

新建一个Executor配置类，顺便把@EnableAsync注解搬到这里来：

@Configuration
@EnableAsync
public class ExecutorConfig {
    /** Set the ThreadPoolExecutor's core pool size. */
    private int corePoolSize = 10;
    /** Set the ThreadPoolExecutor's maximum pool size. */
    private int maxPoolSize = 200;
    /** Set the capacity for the ThreadPoolExecutor's BlockingQueue. */
    private int queueCapacity = 10;
 
    @Bean
    public Executor mySimpleAsync() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(queueCapacity);
        executor.setThreadNamePrefix("MySimpleExecutor-");
        executor.initialize();
        return executor;
    }
 
    @Bean
    public Executor myAsync() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(queueCapacity);
        executor.setThreadNamePrefix("MyExecutor-");
 
        // rejection-policy：当pool已经达到max size的时候，如何处理新任务
        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

这里定义了两个不同的Executor，第二个重新设置了pool已经达到max size时候的处理方法；同时指定了线程名字的前缀。

编写多线程类使用线程池：

@Component
public class AsyncTaskDemo4 {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @Async("mySimpleAsync")
    public Future<String> doTask1() throws InterruptedException{
        logger.info("Task1 started.");
        long start = System.currentTimeMillis();
        Thread.sleep(5000);
        long end = System.currentTimeMillis();
 
        logger.info("Task1 finished, time elapsed: {} ms.", end-start);
 
        return new AsyncResult<>("Task1 accomplished!");
    }
 
    @Async("myAsync")
    public Future<String> doTask2() throws InterruptedException{
        logger.info("Task2 started.");
        long start = System.currentTimeMillis();
        Thread.sleep(3000);
        long end = System.currentTimeMillis();
 
        logger.info("Task2 finished, time elapsed: {} ms.", end-start);
 
        return new AsyncResult<>("Task2 accomplished!");
    }
}

测试结果：

2018-09-07 22:41:44.429  INFO 13640 --- [           main] com.work.spring.thread.TaskTests2        : Started TaskTests2 in 4.179 seconds (JVM running for 5.853)
2018-09-07 22:41:44.525  INFO 13640 --- [   MyExecutor-1] c.w.spring.thread.demo4.AsyncTaskDemo4   : Task2 started.
2018-09-07 22:41:44.525  INFO 13640 --- [impleExecutor-1] c.w.spring.thread.demo4.AsyncTaskDemo4   : Task1 started.
2018-09-07 22:41:47.526  INFO 13640 --- [   MyExecutor-1] c.w.spring.thread.demo4.AsyncTaskDemo4   : Task2 finished, time elapsed: 3001 ms.
2018-09-07 22:41:49.526  INFO 13640 --- [impleExecutor-1] c.w.spring.thread.demo4.AsyncTaskDemo4   : Task1 finished, time elapsed: 5001 ms.
2018-09-07 22:41:50.524  INFO 13640 --- [           main] com.work.spring.thread.TaskTests2        : Task1 result: Task1 accomplished!
2018-09-07 22:41:50.524  INFO 13640 --- [           main] com.work.spring.thread.TaskTests2        : Task2 result: Task2 accomplished!
2018-09-07 22:41:50.524  INFO 13640 --- [           main] com.work.spring.thread.TaskTests2        : All tasks finished.
2018-09-07 22:41:50.528  INFO 13640 --- [       Thread-3] ConfigServletWebServerApplicationContext : Closing org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@747f281: startup date [Fri Sep 07 22:41:40 CST 2018]; root of context hierarchy
2018-09-07 22:41:50.532  INFO 13640 --- [       Thread-3] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'myAsync'

线程名字的前缀变了，可见两个task使用了不同的线程池了。