springboot 多线程执行

Posted 2021-11-03 djq-jone

一.springboot开线程执行异步任务

1.Spring通过任务执行器TaskExecutor，来实现多线程和并发编程，使用ThreadPoolTaskExecutor可实现一个基于线程池的TaskExecutor;

2.异步需要在配置类上面加@EnableAsync 来开启对异步任务的支持在需要异步执行的方法上面加@Async 来声明这个方法是一个需要异步执行的方法;

3.让配置类实现AsyncConfigurer接口，并重写getAsyncExecutor方法，并返回一个ThreasPoolTaskExecutor，就可以获取一个基于线程池的TaskExecutor;

4.@Async 用在方法上，表示这个方法是一个异步的方法，如果用在类上面，表明这个类中的所有方法都是异步的方法。

(1).配置类

 1 package com.yunzhangfang.springboot1.config;
 2 
 3 import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
 4 import org.springframework.context.annotation.ComponentScan;
 5 import org.springframework.context.annotation.Configuration;
 6 import org.springframework.scheduling.annotation.AsyncConfigurer;
 7 import org.springframework.scheduling.annotation.EnableAsync;
 8 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 9 
10 import java.util.concurrent.Executor;
11 
12 @Configuration
13 @ComponentScan("com.yunzhangfang.springboot1.service")
14 @EnableAsync
15 public class ThreadConfig implements AsyncConfigurer {
16     
17     // ThredPoolTaskExcutor的处理流程
18     // 当池子大小小于corePoolSize，就新建线程，并处理请求
19     // 当池子大小等于corePoolSize，把请求放入workQueue中，池子里的空闲线程就去workQueue中取任务并处理
20     // 当workQueue放不下任务时，就新建线程入池，并处理请求，如果池子大小撑到了maximumPoolSize，就用RejectedExecutionHandler来做拒绝处理
21     // 当池子的线程数大于corePoolSize时，多余的线程会等待keepAliveTime长时间，如果无请求可处理就自行销毁
22     @Override
23     public Executor getAsyncExecutor() {
24         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
25         executor.setCorePoolSize(5);
26         executor.setMaxPoolSize(15);
27         executor.setQueueCapacity(25);
28         executor.initialize();
29         return executor;
30     }
31 
32     @Override
33     public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
34         return null;
35     }
36 
37 }

(2).异步任务

 1 package com.yunzhangfang.springboot1.service;
 2 
 3 import org.springframework.scheduling.annotation.Async;
 4 import org.springframework.scheduling.annotation.AsyncResult;
 5 import org.springframework.stereotype.Service;
 6 
 7 import java.util.concurrent.Future;
 8 
 9 @Service
10 public class TaskService {
11 
12     @Async
13     /**
14      * 表明是异步调用
15      * 没有返回值
16      */
17     public void excutVoidTask(int i) {
18         System.out.println("异步执行任务第[" + i + "] 个");
19     }
20 
21     /**
22      * 有返回值
23      * 异常调用
24      *
25      * @param i
26      * @return
27      * @throws InterruptedException
28      */
29     @Async
30     public Future<String> excuteValueTask(int i) throws InterruptedException {
31         Thread.sleep(1000);
32         Future<String> future = new AsyncResult<String>("success is " + i);
33         System.out.println("异步执行任务第[" + i + "] 个");
34         return future;
35     }
36 
37 }

(3).测试异步任务

 1 package com.yunzhangfang.springboot1;
 2 
 3 import com.yunzhangfang.springboot1.service.TaskService;
 4 import org.junit.Test;
 5 import org.junit.runner.RunWith;
 6 import org.springframework.beans.factory.annotation.Autowired;
 7 import org.springframework.boot.test.context.SpringBootTest;
 8 import org.springframework.core.task.TaskRejectedException;
 9 import org.springframework.test.context.junit4.SpringRunner;
10 
11 import java.util.ArrayList;
12 import java.util.List;
13 import java.util.concurrent.ExecutionException;
14 import java.util.concurrent.Future;
15 
16 @RunWith(SpringRunner.class)
17 @SpringBootTest
18 public class Springboot1ApplicationTests {
19 
20     @Autowired
21     private TaskService service;
22 
23     @Test
24     public void contextLoads() {
25     }
26 
27     /**
28      * 没有返回值测试
29      */
30     @Test
31     public void testVoid() {
32         for (int i = 0; i < 20; i++) {
33             service.excutVoidTask(i);
34         }
35         System.out.println("========主线程执行完毕=========");
36     }
37 
38     @Test
39     public void testReturn() throws InterruptedException, ExecutionException {
40         List<Future<String>> lstFuture = new ArrayList<>();// 存放所有的线程，用于获取结果
41         for (int i = 0; i < 100; i++) {
42             while (true) {
43                 try {
44                     // 线程池超过最大线程数时，会抛出TaskRejectedException，则等待1s，直到不抛出异常为止
45                     Future<String> stringFuture = service.excuteValueTask(i);
46                     lstFuture.add(stringFuture);
47                     break;
48                 } catch (TaskRejectedException e) {
49                     System.out.println("线程池满，等待1S。");
50                     Thread.sleep(1000);
51                 }
52             }
53         }
54 
55         // 获取值.get是阻塞式，等待当前线程完成才返回值
56         for (Future<String> future : lstFuture) {
57             System.out.println(future.get());
58         }
59 
60         System.out.println("========主线程执行完毕=========");
61     }
62 
63 }

---

有错误,希望指出,共同进步,天天向上