Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 75dd3daef8 | |||
| c76c902319 |
@@ -1,262 +1 @@
|
|||||||
# 使用 `@Async` 实现线程池
|
https://doc.hutool.cn/pages/ExecutorBuilder
|
||||||
|
|
||||||
```mermaid
|
|
||||||
graph TD
|
|
||||||
A[调用者调用异步方法] --> B[Spring AOP 代理拦截方法调用]
|
|
||||||
B --> C{方法是否为 @Async?}
|
|
||||||
C -->|是| D[封装为 FutureTask 或 Runnable]
|
|
||||||
C -->|否| E[按正常方式执行]
|
|
||||||
D --> F[提交到线程池 TaskExecutor]
|
|
||||||
F --> G[线程池分配一个线程]
|
|
||||||
G --> H[JVM 创建线程并执行任务]
|
|
||||||
H --> I[任务执行完毕,返回结果]
|
|
||||||
I --> J[调用者通过 Future 获取结果]
|
|
||||||
```
|
|
||||||
|
|
||||||
## 前置知识
|
|
||||||
|
|
||||||
### **`@Async`**
|
|
||||||
|
|
||||||
`@Async` 是 Spring 框架提供的一个注解,用于支持**异步方法调用**。它允许我们标记一个方法为异步方法,这样该方法将在**单独的线程**中执行,而不是阻塞当前线程。
|
|
||||||
|
|
||||||
#### 优点:
|
|
||||||
- 提高应用程序的响应速度
|
|
||||||
- 避免长时间运行的操作阻塞主线程
|
|
||||||
- 适用于处理耗时操作、并行任务、异步通知等场景
|
|
||||||
|
|
||||||
#### 使用场景:
|
|
||||||
- 异步发送邮件、短信
|
|
||||||
- 异步日志记录
|
|
||||||
- 异步计算任务
|
|
||||||
- 异步调用第三方服务
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 一、开启异步支持
|
|
||||||
|
|
||||||
要使用 `@Async` 注解,必须在 Spring 应用中开启**异步支持**。可以通过在主类或配置类上添加 `@EnableAsync` 注解来实现。
|
|
||||||
|
|
||||||
```java
|
|
||||||
@Configuration
|
|
||||||
@EnableAsync
|
|
||||||
public class AppConfig {
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
或者在 Spring Boot 主类上添加:
|
|
||||||
|
|
||||||
```java
|
|
||||||
@SpringBootApplication
|
|
||||||
@EnableAsync
|
|
||||||
public class Application {
|
|
||||||
public static void main(String[] args) {
|
|
||||||
SpringApplication.run(Application.class, args);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 二、配置线程池(推荐)
|
|
||||||
|
|
||||||
Spring 提供了多种方式来配置异步任务执行的线程池,推荐使用 `TaskExecutor` 来自定义线程池,以控制并发行为。
|
|
||||||
|
|
||||||
### 1. 使用 `@Bean` 配置自定义线程池
|
|
||||||
|
|
||||||
```java
|
|
||||||
@Configuration
|
|
||||||
@EnableAsync
|
|
||||||
public class AsyncConfig {
|
|
||||||
|
|
||||||
@Bean(name = "customTaskExecutor")
|
|
||||||
public Executor customTaskExecutor() {
|
|
||||||
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
|
||||||
executor.setCorePoolSize(5);
|
|
||||||
executor.setMaxPoolSize(10);
|
|
||||||
executor.setQueueCapacity(100);
|
|
||||||
executor.setThreadNamePrefix("Custom-Async-");
|
|
||||||
executor.initialize();
|
|
||||||
return executor;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
### 2. 指定使用哪个线程池
|
|
||||||
|
|
||||||
可以通过 `@Async("customTaskExecutor")` 指定使用自定义的线程池。
|
|
||||||
|
|
||||||
```java
|
|
||||||
@Service
|
|
||||||
public class MyService {
|
|
||||||
|
|
||||||
@Async("customTaskExecutor")
|
|
||||||
public void asyncMethod() {
|
|
||||||
// 异步执行的操作
|
|
||||||
}
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 三、使用默认线程池
|
|
||||||
|
|
||||||
如果未配置自定义线程池,Spring 会使用默认的 `SimpleAsyncTaskExecutor`,但这个线程池不推荐用于生产环境,因为它**每次都会创建新线程**,容易引发资源耗尽问题。
|
|
||||||
|
|
||||||
可以使用 `@Async` 并不指定线程池名称,Spring 会自动使用默认的线程池(需确保 `@EnableAsync` 已开启):
|
|
||||||
|
|
||||||
```java
|
|
||||||
@Service
|
|
||||||
public class MyService {
|
|
||||||
|
|
||||||
@Async
|
|
||||||
public void defaultAsyncMethod() {
|
|
||||||
// 默认线程池执行
|
|
||||||
}
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 四、异步方法调用注意事项
|
|
||||||
|
|
||||||
- `@Async` 注解的方法不能和调用它的方法在同一个类中,否则不会生效。因为 Spring 的 AOP 机制需要通过代理来实现异步调用,若在同一个类中调用,代理无法生效。
|
|
||||||
- 异步方法不能返回 `void`,否则无法捕获异常。**建议返回 `Future<T>` 或 `CompletableFuture<T>`**。
|
|
||||||
- 异步方法的异常处理:如果希望捕获异步方法中的异常,可以在方法中使用 `try-catch`,或者通过 `Future` 返回异常信息。
|
|
||||||
- 如果异步方法中涉及事务控制,需注意事务的传播机制。
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 五、示例代码
|
|
||||||
|
|
||||||
### 示例 1:简单异步方法
|
|
||||||
|
|
||||||
```java
|
|
||||||
@Service
|
|
||||||
public class AsyncService {
|
|
||||||
|
|
||||||
@Async
|
|
||||||
public void sendNotification(String message) {
|
|
||||||
System.out.println("Sending notification: " + message);
|
|
||||||
// 模拟耗时操作
|
|
||||||
try {
|
|
||||||
Thread.sleep(1000);
|
|
||||||
} catch (InterruptedException e) {
|
|
||||||
e.printStackTrace();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
### 示例 2:使用自定义线程池并返回 Future
|
|
||||||
|
|
||||||
```java
|
|
||||||
@Service
|
|
||||||
public class AsyncService {
|
|
||||||
|
|
||||||
@Async("customTaskExecutor")
|
|
||||||
public Future<String> asyncTask(String input) {
|
|
||||||
try {
|
|
||||||
Thread.sleep(2000);
|
|
||||||
return new AsyncResult<>("Processed: " + input);
|
|
||||||
} catch (InterruptedException e) {
|
|
||||||
return new AsyncResult<>("Error occurred");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
在调用处使用:
|
|
||||||
|
|
||||||
```java
|
|
||||||
@Autowired
|
|
||||||
private AsyncService asyncService;
|
|
||||||
|
|
||||||
public void callAsyncTask() {
|
|
||||||
Future<String> result = asyncService.asyncTask("test");
|
|
||||||
try {
|
|
||||||
System.out.println(result.get());
|
|
||||||
} catch (Exception e) {
|
|
||||||
e.printStackTrace();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 六、线程池配置属性说明
|
|
||||||
|
|
||||||
| 属性名 | 描述 |
|
|
||||||
|-----------------------|--------------------------------------------|
|
|
||||||
| `corePoolSize` | 核心线程数(常驻线程数量) |
|
|
||||||
| `maxPoolSize` | 最大线程数(允许创建的最大线程数) |
|
|
||||||
| `queueCapacity` | 任务队列容量(缓存未被立即执行的任务) |
|
|
||||||
| `keepAliveSeconds` | 非核心线程的空闲时间(超过后会被销毁) |
|
|
||||||
| `threadNamePrefix` | 线程名称前缀(用于调试和日志区分) |
|
|
||||||
| `allowCoreThreadTimeOut` | 是否允许核心线程超时(默认 false) |
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 七、异步方法异常处理
|
|
||||||
|
|
||||||
由于异步方法在独立线程中执行,不能直接抛出异常。可使用 `Future` 或 `CompletableFuture` 来捕获和处理异常。
|
|
||||||
|
|
||||||
### 使用 `Future`
|
|
||||||
|
|
||||||
```java
|
|
||||||
@Async
|
|
||||||
public Future<String> asyncMethod() {
|
|
||||||
try {
|
|
||||||
// …
|
|
||||||
return new AsyncResult<>("Success");
|
|
||||||
} catch (Exception e) {
|
|
||||||
return new AsyncResult<>("Error: " + e.getMessage());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
### 使用 `CompletableFuture`
|
|
||||||
|
|
||||||
```java
|
|
||||||
@Async
|
|
||||||
public CompletableFuture<String> asyncMethod() {
|
|
||||||
return CompletableFuture.supplyAsync(() -> {
|
|
||||||
try {
|
|
||||||
// …
|
|
||||||
return "Success";
|
|
||||||
} catch (Exception e) {
|
|
||||||
throw new RuntimeException(e);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 八、常见问题与排查
|
|
||||||
|
|
||||||
| 问题 | 解决方案 |
|
|
||||||
|------|----------|
|
|
||||||
| `@Async` 方法未执行 | 检查是否开启 `@EnableAsync`,是否在不同类中调用 |
|
|
||||||
| 线程池配置无效 | 确保 `@Bean` 配置正确,且 `@Async` 指定了正确的 bean 名称 |
|
|
||||||
| 线程池资源耗尽 | 调整线程池大小和队列容量,增加 `queueCapacity` |
|
|
||||||
| 异常未被捕获 | 使用 `Future` 或 `CompletableFuture` 返回结果,避免 `void` 返回 |
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 九、相关类与接口
|
|
||||||
|
|
||||||
- `org.springframework.scheduling.annotation.Async`
|
|
||||||
- `org.springframework.scheduling.annotation.EnableAsync`
|
|
||||||
- `org.springframework.core.task.AsyncTaskExecutor`
|
|
||||||
- `org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor`
|
|
||||||
- `org.springframework.util.concurrent.AsyncTask`
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 十、最佳实践
|
|
||||||
|
|
||||||
- 为不同业务场景配置不同的线程池
|
|
||||||
- 使用 `CompletableFuture` 来处理异步任务中的返回结果和异常
|
|
||||||
- 避免在异步方法中使用共享变量,防止线程安全问题
|
|
||||||
- 控制线程池的大小和任务队列,防止资源浪费或阻塞
|
|
||||||
- 结合 `@Scheduled` 注解实现定时异步任务
|
|
||||||
@@ -23,6 +23,12 @@
|
|||||||
</parent>
|
</parent>
|
||||||
|
|
||||||
<dependencies>
|
<dependencies>
|
||||||
|
<dependency>
|
||||||
|
<groupId>cn.hutool</groupId>
|
||||||
|
<artifactId>hutool-core</artifactId>
|
||||||
|
<version>5.8.31</version>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>org.springframework.boot</groupId>
|
<groupId>org.springframework.boot</groupId>
|
||||||
<artifactId>spring-boot-starter-web</artifactId>
|
<artifactId>spring-boot-starter-web</artifactId>
|
||||||
|
|||||||
@@ -1,11 +1,42 @@
|
|||||||
package cn.hezhaohui.threadpool;
|
package cn.hezhaohui.threadpool;
|
||||||
|
|
||||||
import org.springframework.boot.SpringApplication;
|
import cn.hutool.core.thread.ExecutorBuilder;
|
||||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
|
||||||
@SpringBootApplication
|
import java.util.concurrent.ArrayBlockingQueue;
|
||||||
|
import java.util.concurrent.RejectedExecutionHandler;
|
||||||
|
import java.util.concurrent.ThreadPoolExecutor;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
|
@Slf4j
|
||||||
public class ThreadPoolApplication {
|
public class ThreadPoolApplication {
|
||||||
public static void main(String[] args) {
|
public static void main(String[] args) {
|
||||||
SpringApplication.run(ThreadPoolApplication.class, args);
|
ExecutorBuilder builder = new ExecutorBuilder();
|
||||||
|
ThreadPoolExecutor executor = builder
|
||||||
|
.setCorePoolSize(5)
|
||||||
|
.setMaxPoolSize(10)
|
||||||
|
.setKeepAliveTime(0)
|
||||||
|
.setHandler(new RejectedExecutionHandler(){
|
||||||
|
@Override
|
||||||
|
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
|
||||||
|
log.info(r.toString());
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.setWorkQueue(new ArrayBlockingQueue<>(100))
|
||||||
|
.build();
|
||||||
|
|
||||||
|
for (int i = 1; i <= 1000; i++) {
|
||||||
|
executor.execute(() -> {
|
||||||
|
log.info("Hello");
|
||||||
|
try {
|
||||||
|
Thread.sleep(1000);
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
throw new RuntimeException(e);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
executor.shutdown();
|
||||||
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,29 +0,0 @@
|
|||||||
package cn.hezhaohui.threadpool.conifg;
|
|
||||||
|
|
||||||
import org.springframework.context.annotation.Bean;
|
|
||||||
import org.springframework.context.annotation.Configuration;
|
|
||||||
import org.springframework.scheduling.annotation.EnableAsync;
|
|
||||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
|
||||||
|
|
||||||
import java.util.concurrent.Executor;
|
|
||||||
import java.util.concurrent.ThreadPoolExecutor;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 异步配置类
|
|
||||||
*/
|
|
||||||
@Configuration
|
|
||||||
@EnableAsync
|
|
||||||
public class AsyncConfig {
|
|
||||||
|
|
||||||
@Bean(name = "customThreadPool")
|
|
||||||
public Executor customThreadPool() {
|
|
||||||
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
|
||||||
executor.setCorePoolSize(5);
|
|
||||||
executor.setMaxPoolSize(10);
|
|
||||||
executor.setQueueCapacity(100);
|
|
||||||
executor.setThreadNamePrefix("Custom-Async-");
|
|
||||||
executor.initialize();
|
|
||||||
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
|
|
||||||
return executor;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,45 +0,0 @@
|
|||||||
package cn.hezhaohui.threadpool.service;
|
|
||||||
|
|
||||||
import lombok.extern.slf4j.Slf4j;
|
|
||||||
import org.springframework.scheduling.annotation.Async;
|
|
||||||
import org.springframework.stereotype.Service;
|
|
||||||
|
|
||||||
import java.util.Random;
|
|
||||||
import java.util.UUID;
|
|
||||||
import java.util.concurrent.CountDownLatch;
|
|
||||||
|
|
||||||
@Service
|
|
||||||
@Slf4j
|
|
||||||
public class AsyncService {
|
|
||||||
|
|
||||||
@Async("customThreadPool")
|
|
||||||
public void asyncPrint() {
|
|
||||||
String id = UUID.randomUUID().toString().substring(0, 5);
|
|
||||||
int sleepSeconds = new Random().nextInt(15);
|
|
||||||
log.info("[Thread {}] Called, waiting {} seconds", id, sleepSeconds);
|
|
||||||
try {
|
|
||||||
Thread.sleep(sleepSeconds * 1000L);
|
|
||||||
log.info("[Thread {}] Wait completed after {} seconds", id, sleepSeconds);
|
|
||||||
} catch (InterruptedException e) {
|
|
||||||
log.error("[Thread {}] Interrupted during wait", id, e);
|
|
||||||
Thread.currentThread().interrupt();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@Async("customThreadPool")
|
|
||||||
public void asyncPrint(CountDownLatch latch) {
|
|
||||||
String id = UUID.randomUUID().toString().substring(0, 5);
|
|
||||||
int sleepSeconds = new Random().nextInt(15);
|
|
||||||
log.info("[Thread {}] ⚠️ Called, waiting {} seconds", id, sleepSeconds);
|
|
||||||
try {
|
|
||||||
Thread.sleep(sleepSeconds * 1000L);
|
|
||||||
log.info("[Thread {}] ✅ Wait completed after {} seconds", id, sleepSeconds);
|
|
||||||
} catch (InterruptedException e) {
|
|
||||||
log.error("[Thread {}] Interrupted during wait", id, e);
|
|
||||||
Thread.currentThread().interrupt();
|
|
||||||
} finally {
|
|
||||||
latch.countDown();
|
|
||||||
log.info("⭐ {}", latch.getCount());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,29 +0,0 @@
|
|||||||
package cn.hezhaohui.threadpool.service;
|
|
||||||
|
|
||||||
import jakarta.annotation.Resource;
|
|
||||||
import lombok.extern.slf4j.Slf4j;
|
|
||||||
import org.junit.jupiter.api.Test;
|
|
||||||
import org.springframework.boot.test.context.SpringBootTest;
|
|
||||||
|
|
||||||
import java.util.concurrent.CountDownLatch;
|
|
||||||
|
|
||||||
import static org.junit.jupiter.api.Assertions.*;
|
|
||||||
|
|
||||||
@SpringBootTest
|
|
||||||
@Slf4j
|
|
||||||
class AsyncServiceTest {
|
|
||||||
|
|
||||||
@Resource
|
|
||||||
private AsyncService asyncService;
|
|
||||||
|
|
||||||
@Test
|
|
||||||
void asyncPrint() throws InterruptedException {
|
|
||||||
CountDownLatch latch = new CountDownLatch(100);
|
|
||||||
for (int i = 0; i < 100; i++) {
|
|
||||||
asyncService.asyncPrint(latch);
|
|
||||||
}
|
|
||||||
log.info("\n[+++ All Tasks Commited +++]\n");
|
|
||||||
latch.await();
|
|
||||||
log.info("\n[=== All Tasks Completed ===]\n");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Reference in New Issue
Block a user