Java异步模块案例如何复用

本文目录导读：

1. 线程池复用（最基础且重要）
2. 通过依赖注入复用异步结果
3. 异步模块封装为可复用的组件
4. 异步事件复用（使用Spring Event）
5. 异步任务抽象工厂
6. 关键复用原则

在Java中复用异步模块,主要目标是在多个场景或请求间共享异步资源（如线程池、异步任务结果等），避免重复创建线程池、连接池等开销，以下是核心复用策略和案例：

线程池复用（最基础且重要）

错误示范：每个异步方法都创建线程池

public class BadAsyncService {
    public CompletableFuture<String> doTask() {
        ExecutorService executor = Executors.newFixedThreadPool(10);  // 每次调用都创建！
        return CompletableFuture.supplyAsync(() -> "result", executor);
    }
}

正确复用：使用单例线程池

@Component
public class AsyncExecutorConfig {
    @Bean("asyncExecutor")
    public Executor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(500);
        executor.setThreadNamePrefix("async-pool-");
        executor.initialize();
        return executor;
    }
}
@Service
public class AsyncService {
    @Autowired
    @Qualifier("asyncExecutor")
    private Executor asyncExecutor;
    @Async("asyncExecutor")  // Spring @Async 也可以指定线程池
    public CompletableFuture<String> method1() {
        // 复用同一线程池
        return CompletableFuture.completedFuture("result1");
    }
    public CompletableFuture<String> method2() {
        return CompletableFuture.supplyAsync(() -> "result2", asyncExecutor);
    }
}

通过依赖注入复用异步结果

@Component
public class AsyncResultCache {
    private final Map<String, CompletableFuture<UserData>> cache = new ConcurrentHashMap<>();
    @Autowired
    private UserService userService;
    @Autowired
    private Executor asyncExecutor;
    public CompletableFuture<UserData> getUserData(String userId) {
        // 复用已存在的异步任务（防止重复请求）
        return cache.computeIfAbsent(userId, id -> 
            CompletableFuture.supplyAsync(() -> 
                userService.fetchUserFromRemote(userId), asyncExecutor
            )
        );
    }
    @Scheduled(fixedDelay = 60000)
    public void cleanExpiredCache() {
        // 定期清理过期的缓存
        cache.entrySet().removeIf(entry -> entry.getValue().isDone());
    }
}

异步模块封装为可复用的组件

// 通用的异步批处理处理器
@Component
public class AsyncBatchProcessor {
    private final Executor executor;
    private final int batchSize;
    public AsyncBatchProcessor(@Value("${batch.size:100}") int batchSize,
                               @Qualifier("asyncExecutor") Executor executor) {
        this.batchSize = batchSize;
        this.executor = executor;
    }
    public <T, R> CompletableFuture<List<R>> processBatch(List<T> items,
                                                          Function<List<T>, List<R>> processor) {
        List<CompletableFuture<List<R>>> futures = new ArrayList<>();
        // 分批提交异步任务
        for (int i = 0; i < items.size(); i += batchSize) {
            int end = Math.min(i + batchSize, items.size());
            List<T> batch = items.subList(i, end);
            CompletableFuture<List<R>> future = CompletableFuture.supplyAsync(
                () -> processor.apply(batch), executor);
            futures.add(future);
        }
        // 合并所有批次的结果
        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                .thenApply(v -> futures.stream()
                    .map(CompletableFuture::join)
                    .flatMap(List::stream)
                    .collect(Collectors.toList()));
    }
}
// 使用示例
@Service
public class DataService {
    @Autowired
    private AsyncBatchProcessor batchProcessor;
    public CompletableFuture<List<User>> processUsers(List<Integer> ids) {
        return batchProcessor.processBatch(ids, this::fetchUsersFromDb);
    }
    private List<User> fetchUsersFromDb(List<Integer> batchIds) {
        // 实际的数据库查询
        return userRepository.findByIdIn(batchIds);
    }
}

异步事件复用（使用Spring Event）

// 定义一个可复用的异步事件发布器
@Component
public class AsyncEventPublisher {
    @Autowired
    private ApplicationEventPublisher publisher;
    @Async("asyncExecutor")
    public void publishAsync(Object event) {
        publisher.publishEvent(event);
    }
}
// 监听器可以复用事件处理逻辑
@Component
public class UserEventHandler {
    @Async("asyncExecutor")
    @EventListener
    public void handleUserCreated(UserCreatedEvent event) {
        // 异步处理用户创建后的逻辑
        sendWelcomeEmail(event.getUserId());
        initializeUserResources(event.getUserId());
    }
    @Async("asyncExecutor")
    @EventListener
    public void handleUserUpdated(UserUpdatedEvent event) {
        // 同样的异步处理器，复用线程池
        syncToSearchEngine(event.getUserId());
        invalidateCache(event.getUserId());
    }
}

异步任务抽象工厂

// 定义通用的异步任务接口
public interface AsyncTask<T> {
    CompletableFuture<T> execute();
}
// 抽象工厂，复用异步执行骨架
public abstract class AbstractAsyncTaskExecutor {
    @Autowired
    private Executor asyncExecutor;
    public <T> CompletableFuture<T> execute(AsyncTask<T> task) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                return task.execute().get();
            } catch (Exception e) {
                throw new RuntimeException("Async task failed", e);
            }
        }, asyncExecutor);
    }
    protected <T> CompletableFuture<Void> executeParallel(List<AsyncTask<T>> tasks) {
        List<CompletableFuture<T>> futures = tasks.stream()
            .map(task -> CompletableFuture.supplyAsync(() -> {
                try {
                    return task.execute().get();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }, asyncExecutor))
            .collect(Collectors.toList());
        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
    }
}
// 具体任务实现
@Component
public class EmailAsyncTask implements AsyncTask<Boolean> {
    @Override
    public CompletableFuture<Boolean> execute() {
        // 发送邮件的具体逻辑
        return CompletableFuture.completedFuture(true);
    }
}

关键复用原则

通过以上方式,可以有效减少资源消耗，提升系统吞吐量，同时保持代码的模块化和可维护性。