Java线程池的实现--Executor、ThreadPoolTaskExecutor、@Async的使用

╰半夏微凉° 2023-06-23 07:56 5阅读 0赞

一、为什么要使用线程池  
当我们需要的并发执行线程数量很多时，且每个线程执行很短的时间就结束了，这样，我们频繁的创建、销毁线程就大大降低了工作效率（创建和销毁线程需要时间、资源）。java中的线程池可以达到这样的效果：一个线程执行完任务之后，继续去执行下一个任务，不被销毁，这样线程利用率提高了。

二、Jdk 1.5后线程池  
    //【固定的线程池】定义进程池并指定其大小    
    ExecutorService threadPool = Executors.newFixedThreadPool(3);

for(int i=1;i<=10;i++)\{  
            final int task = i;  
            threadPool.execute(new Runnable()\{  
                @Override  
                public void run() \{  
                    for(int j=1;j<=2;j++)\{  
                        try \{  
                            Thread.sleep(20);  
                        \} catch (InterruptedException e) \{  
                            e.printStackTrace();  
                        \}  
                        System.out.println(Thread.currentThread().getName() + " is looping of " + j + " for  task of " + task);  
                    \}  
                \}  
            \});  
        \}  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
java.util.concurrent.Executor接口表示线程池，它的execute(Runnable task)用来执行Runnable类型的任务。Executor的子接口ExecutorService中声明了管理线程池的一些方法，比如关闭线程池的shutdown()等。Executors类中包含了一些静态方法，生成各种类型的线程池ExecutorService实例。

三、Spring异步线程池  
异步调用

在解释异步调用之前，我们先来看同步调用的定义；同步就是整个处理过程顺序执行，当各个过程都执行完毕，并返回结果。 异步调用则是只是发送了调用的指令，调用者无需等待被调用的方法完全执行完毕；而是继续执行下面的流程。

1、TaskExecutor  
TaskExecutor是Spring异步线程池的接口类，其实质是java.util.concurrent.Executor。

Spring 已经实现的异常线程池：

SimpleAsyncTaskExecutor：不是真的线程池，每次调用都会创建一个新的线程。  
SyncTaskExecutor：这个类没有实现异步调用，只是一个同步操作。只适用于不需要多线程的地方。  
ConcurrentTaskExecutor：Executor的适配类，不推荐使用。如果ThreadPoolTaskExecutor不满足要求时，才用考虑使用这个类。  
SimpleThreadPoolTaskExecutor：是Quartz的SimpleThreadPool的类。线程池同时被quartz和非quartz使用，才需要使用此类。  
ThreadPoolTaskExecutor：最常使用，推荐。 其实质是对java.util.concurrent.ThreadPoolExecutor的包装。  
ThreadPoolTaskExecutor的使用：

//spring.xml配置  
    
<bean id="taskExecutor"    
    class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">    
        
    <property name="corePoolSize" value="10" />    
        
    <property name="maxPoolSize" value="100" />    
        
    <property name="queueCapacity" value="1000" />    
        
    <property name="keepAliveSeconds" value="300" />    
        
    <property name="rejectedExecutionHandler">    
        <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" />    
    </property>    
</bean>

//从池中获取线程执行任务  
public class ThreadPoolTest \{    
    @Autowired    
    private TaskExecutor taskExecutor;// 线程池

// 将创建的线程添加到线程池中    
    public void test() throws Exception \{    
        for (int i = 0; i < 10; i++) \{    
            this.taskExecutor.execute(new AppContentDataPushThread());    
        \}    
    \}

class AppContentDataPushThread implements Runnable \{

public AppContentDataPushThread() \{    
        \}

@Override    
        public void run() \{    
            System.out.println("执行线程");    
        \}    
    \}    
\}    
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
27  
28  
29  
30  
31  
32  
33  
34  
35  
36  
37  
38  
39  
40  
41  
42  
2、@Async  
2.1、在Spring中启用@Async  
基于注解配置

@Configuration    
@EnableAsync    
public class SpringAsyncConfig \{ ... \}    
1  
2  
3  
基于xml配置

<task:executor id="myexecutor"pool-size="5"/>     
<task:annotation-driven executor="myexecutor"/>    
1  
2  
2.2、使用  
@Component  
public class AsyncDemo \{  
    private static final Logger log = LoggerFactory.getLogger(AsyncDemo.class);

/\*\*  
     \* 最简单的异步调用，返回值为void  
     \*/  
    @Async  
    public void asyncInvokeSimplest() \{  
        log.info("asyncSimplest");  
    \}

/\*\*  
     \* 带参数的异步调用 异步方法可以传入参数  
     \*   
     \* @param s  
     \*/  
    @Async  
    public void asyncInvokeWithParameter(String s) \{  
        log.info("asyncInvokeWithParameter, parementer=\{\}", s);  
    \}

/\*\*  
     \* 异常调用返回Future  
     \*   
     \* @param i  
     \* @return  
     \*/  
    @Async  
    public Future<String> asyncInvokeReturnFuture(int i) \{  
        log.info("asyncInvokeReturnFuture, parementer=\{\}", i);  
        Future<String> future;  
        try \{  
            Thread.sleep(1000 \* 1);  
            future = new AsyncResult<String>("success:" + i);  
        \} catch (InterruptedException e) \{  
            future = new AsyncResult<String>("error");  
        \}  
        return future;  
    \}  
\}

//调用  
asyncDemo.asyncInvokeSimplest();  
asyncDemo.asyncInvokeWithException("test");  
Future<String> future = asyncDemo.asyncInvokeReturnFuture(100);  
System.out.println(future.get());