Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 wonder
|
61860a3190 | ||
|
wonder
|
fc9972df3b | ||
|
wonder
|
770527193e |
@@ -1,239 +1,262 @@
|
||||
# spring-boot-threadpool-demo
|
||||
# 使用 `@Async` 实现线程池
|
||||
|
||||
> 📚 适用于 SpringBoot 的多种线程池实现
|
||||
|
||||
**注意：Main 分支不包含具体实现，具体实现在项目的不同分支上**
|
||||
|
||||
## 项目结构
|
||||
```
|
||||
spring-boot-threadpool-demo/
|
||||
├── src/
|
||||
│ ├── main/
|
||||
│ │ ├── java/cn/hezhaohui/threadpool/
|
||||
│ │ │ ├── ThreadpoolApplication.java
|
||||
│ │ │ ├── config/
|
||||
│ │ │ ├── service/
|
||||
│ │ │ ├── controller/
|
||||
│ │ │ └── model/
|
||||
│ └── resources/
|
||||
└── README.md
|
||||
```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 获取结果]
|
||||
```
|
||||
|
||||
## 分支大纲
|
||||
## 前置知识
|
||||
|
||||
### 分支 1: `basic-async-config`
|
||||
**主题：基础 @Async 注解方式**
|
||||
- 实现简单的异步任务执行
|
||||
- 配置基本的线程池参数
|
||||
- 测试异步方法调用
|
||||
- 包含线程安全性验证
|
||||
### **`@Async`**
|
||||
|
||||
### 分支 2: `custom-executor-config`
|
||||
**主题：自定义 Executor 配置**
|
||||
- 使用 ThreadPoolTaskExecutor 自定义配置
|
||||
- 配置核心线程数、最大线程数、队列容量
|
||||
- 设置线程名称前缀和拒绝策略
|
||||
- 添加线程池监控和管理
|
||||
`@Async` 是 Spring 框架提供的一个注解，用于支持**异步方法调用**。它允许我们标记一个方法为异步方法，这样该方法将在**单独的线程**中执行，而不是阻塞当前线程。
|
||||
|
||||
### 分支 3: `application-properties-config`
|
||||
**主题：配置文件驱动方式**
|
||||
- 使用 application.yml 配置线程池
|
||||
- 支持不同环境的配置切换
|
||||
- 动态配置线程池参数
|
||||
- 配置文件验证和异常处理
|
||||
#### 优点：
|
||||
- 提高应用程序的响应速度
|
||||
- 避免长时间运行的操作阻塞主线程
|
||||
- 适用于处理耗时操作、并行任务、异步通知等场景
|
||||
|
||||
### 分支 4: `completable-future-config`
|
||||
**主题：CompletableFuture 方式**
|
||||
- 使用 CompletableFuture 实现异步编程
|
||||
- 异步任务链式调用
|
||||
- 异常处理和结果聚合
|
||||
- 异步回调机制实现
|
||||
#### 使用场景：
|
||||
- 异步发送邮件、短信
|
||||
- 异步日志记录
|
||||
- 异步计算任务
|
||||
- 异步调用第三方服务
|
||||
|
||||
### 分支 5: `scheduled-task-config`
|
||||
**主题：定时任务线程池**
|
||||
- 配置定时任务线程池
|
||||
- 定时任务执行示例
|
||||
- 任务调度和管理
|
||||
- 定时任务与异步任务结合
|
||||
---
|
||||
|
||||
### 分支 6: `advanced-monitoring-config`
|
||||
**主题：高级监控和管理**
|
||||
- 线程池状态监控
|
||||
- 自定义线程池管理器
|
||||
- 性能指标收集
|
||||
- 健康检查和告警机制
|
||||
## 一、开启异步支持
|
||||
|
||||
### 分支 7: `multiple-pool-config`
|
||||
**主题：多线程池配置**
|
||||
- 不同业务场景的线程池隔离
|
||||
- 配置多个不同的线程池
|
||||
- 线程池路由和选择机制
|
||||
- 资源隔离和性能优化
|
||||
要使用 `@Async` 注解，必须在 Spring 应用中开启**异步支持**。可以通过在主类或配置类上添加 `@EnableAsync` 注解来实现。
|
||||
|
||||
### 分支 8: `error-handling-config`
|
||||
**主题：异常处理和容错**
|
||||
- 异步任务异常处理
|
||||
- 重试机制实现
|
||||
- 降级策略
|
||||
- 异常日志记录和监控
|
||||
|
||||
## 各分支详细实现内容
|
||||
|
||||
### 分支 1: basic-async-config
|
||||
```java
|
||||
// 配置类
|
||||
@Configuration
|
||||
@EnableAsync
|
||||
public class AsyncConfig { }
|
||||
|
||||
// 服务类
|
||||
@Service
|
||||
public class AsyncService {
|
||||
@Async
|
||||
public void simpleAsyncTask() { }
|
||||
public class AppConfig {
|
||||
}
|
||||
|
||||
// 测试类
|
||||
@SpringBootTest
|
||||
public class AsyncTest { }
|
||||
```
|
||||
|
||||
### 分支 2: custom-executor-config
|
||||
或者在 Spring Boot 主类上添加：
|
||||
|
||||
```java
|
||||
// 配置类
|
||||
@Configuration
|
||||
@SpringBootApplication
|
||||
@EnableAsync
|
||||
public class CustomAsyncConfig {
|
||||
@Bean("customExecutor")
|
||||
public Executor customExecutor() { }
|
||||
}
|
||||
|
||||
// 服务类
|
||||
@Service
|
||||
public class CustomAsyncService {
|
||||
@Async("customExecutor")
|
||||
public void customAsyncTask() { }
|
||||
}
|
||||
```
|
||||
|
||||
### 分支 3: application-properties-config
|
||||
```yaml
|
||||
# application.yml
|
||||
spring:
|
||||
task:
|
||||
execution:
|
||||
pool:
|
||||
core-size: 5
|
||||
max-size: 10
|
||||
queue-capacity: 100
|
||||
```
|
||||
|
||||
### 分支 4: completable-future-config
|
||||
```java
|
||||
// 服务类
|
||||
@Service
|
||||
public class CompletableFutureService {
|
||||
public CompletableFuture<String> asyncMethod() {
|
||||
return CompletableFuture.supplyAsync(() -> "result");
|
||||
public class Application {
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(Application.class, args);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 分支 5: scheduled-task-config
|
||||
---
|
||||
|
||||
## 二、配置线程池（推荐）
|
||||
|
||||
Spring 提供了多种方式来配置异步任务执行的线程池，推荐使用 `TaskExecutor` 来自定义线程池，以控制并发行为。
|
||||
|
||||
### 1. 使用 `@Bean` 配置自定义线程池
|
||||
|
||||
```java
|
||||
// 配置类
|
||||
@Configuration
|
||||
@EnableScheduling
|
||||
public class ScheduledConfig {
|
||||
@Bean("scheduledExecutor")
|
||||
public Executor scheduledExecutor() { }
|
||||
}
|
||||
|
||||
// 定时任务
|
||||
@Component
|
||||
public class ScheduledTask {
|
||||
@Scheduled(fixedRate = 5000)
|
||||
@Async("scheduledExecutor")
|
||||
public void scheduledTask() { }
|
||||
}
|
||||
```
|
||||
|
||||
### 分支 6: advanced-monitoring-config
|
||||
```java
|
||||
// 线程池监控
|
||||
@Component
|
||||
public class ThreadPoolMonitor {
|
||||
public void monitorThreadPool() { }
|
||||
}
|
||||
|
||||
// 健康检查
|
||||
@RestController
|
||||
public class ThreadPoolHealthController { }
|
||||
```
|
||||
|
||||
### 分支 7: multiple-pool-config
|
||||
```java
|
||||
// 多个线程池配置
|
||||
@Configuration
|
||||
@EnableAsync
|
||||
public class MultiplePoolConfig {
|
||||
@Bean("businessExecutor")
|
||||
public Executor businessExecutor() { }
|
||||
|
||||
@Bean("ioExecutor")
|
||||
public Executor ioExecutor() { }
|
||||
}
|
||||
public class AsyncConfig {
|
||||
|
||||
// 服务类使用不同线程池
|
||||
@Service
|
||||
public class MultiPoolService {
|
||||
@Async("businessExecutor")
|
||||
public void businessTask() { }
|
||||
|
||||
@Async("ioExecutor")
|
||||
public void ioTask() { }
|
||||
@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;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 分支 8: error-handling-config
|
||||
### 2. 指定使用哪个线程池
|
||||
|
||||
可以通过 `@Async("customTaskExecutor")` 指定使用自定义的线程池。
|
||||
|
||||
```java
|
||||
// 异常处理配置
|
||||
@Service
|
||||
public class ErrorHandlingService {
|
||||
@Async("errorHandlingExecutor")
|
||||
public void taskWithErrorHandling() { }
|
||||
}
|
||||
public class MyService {
|
||||
|
||||
// 全局异常处理器
|
||||
@RestControllerAdvice
|
||||
public class AsyncExceptionHandler { }
|
||||
@Async("customTaskExecutor")
|
||||
public void asyncMethod() {
|
||||
// 异步执行的操作
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 测试用例设计
|
||||
---
|
||||
|
||||
### 每个分支都需要包含的测试：
|
||||
1. **线程池配置验证**
|
||||
2. **异步任务执行验证**
|
||||
3. **性能基准测试**
|
||||
4. **异常处理测试**
|
||||
5. **资源使用监控**
|
||||
## 三、使用默认线程池
|
||||
|
||||
## 部署和监控
|
||||
如果未配置自定义线程池，Spring 会使用默认的 `SimpleAsyncTaskExecutor`，但这个线程池不推荐用于生产环境，因为它**每次都会创建新线程**，容易引发资源耗尽问题。
|
||||
|
||||
### 分支 9: `monitoring-deployment`
|
||||
**主题：部署和监控配置**
|
||||
- Docker 部署配置
|
||||
- Prometheus 监控集成
|
||||
- Grafana 可视化
|
||||
- 日志收集和分析
|
||||
可以使用 `@Async` 并不指定线程池名称，Spring 会自动使用默认的线程池（需确保 `@EnableAsync` 已开启）：
|
||||
|
||||
## 文档和说明
|
||||
```java
|
||||
@Service
|
||||
public class MyService {
|
||||
|
||||
### README.md 包含：
|
||||
1. 各分支功能说明
|
||||
2. 配置参数说明
|
||||
3. 使用示例
|
||||
4. 性能测试结果
|
||||
5. 最佳实践建议
|
||||
@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` 注解实现定时异步任务
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
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());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
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