在Spring Boot中实现长耗时API请求的异步执行与取消策略（耗时.请求.取消.策略.执行...）

wufei123 发布于 2025-09-17 阅读（11）

在Spring Boot中实现长耗时API请求的异步执行与取消策略

本文探讨了在Spring Boot应用中如何处理长时间运行的API请求，并实现其动态取消。针对同步阻塞的问题，我们将介绍如何通过异步执行任务、维护任务状态以及利用Java的并发机制，实现对正在执行的API请求进行中断或取消，从而提升用户体验和系统响应性。1. 问题背景与挑战

在传统的spring boot api设计中，如果一个@postmapping或@getmapping方法内部包含了耗时较长的业务逻辑（例如，复杂的计算、大量数据处理、调用外部慢速服务等），并且这些逻辑是同步执行的，那么会带来以下问题：

  1. API阻塞： 客户端请求会一直等待，直到服务器端方法执行完毕并返回结果，导致用户体验不佳。
  2. 资源占用： 服务器线程被长时间占用，降低了服务器处理并发请求的能力。
  3. 无法取消： 一旦请求发出，服务器端任务就无法被外部中断或取消，即使客户端不再需要结果。
  4. 状态管理困难： 对于多个同时发起的长耗时请求，难以追踪和管理它们的执行状态。

原始代码示例中，一个简单的for循环在API方法内部执行，这正是上述问题的典型体现。当多个带有不同timeToRun参数的请求同时到达时，如果用户希望取消其中一些请求的执行，现有的同步模型无法提供支持。

2. 核心思路：异步执行与任务管理

要解决上述问题，核心思想是将耗时操作从主请求线程中分离出来，进行异步执行，并提供一个机制来追踪这些异步任务，以便在需要时进行取消。

具体实现步骤包括：

  1. 异步化任务： 使用Spring的异步执行能力将耗时逻辑放入独立的线程池中执行。
  2. 任务标识与存储： 为每个异步任务生成一个唯一的标识符，并将其对应的Future对象存储起来，以便后续引用。
  3. 取消机制： 提供一个API接口，允许客户端通过任务标识符来请求取消正在执行的任务。
  4. 任务内部响应中断： 异步任务的执行逻辑需要周期性地检查中断状态，并根据中断信号进行优雅地退出。
3. 实现步骤与示例 3.1 配置异步任务执行器

Spring Boot通过@EnableAsync注解和TaskExecutor来支持异步方法。为了更好地控制线程池行为，我们通常会自定义一个ThreadPoolTaskExecutor。

// src/main/java/com/example/async/AsyncConfig.java
package com.example.async;

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;

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "taskExecutor")
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5); // 核心线程数
        executor.setMaxPoolSize(10); // 最大线程数
        executor.setQueueCapacity(25); // 队列容量
        executor.setThreadNamePrefix("AsyncTask-"); // 线程名前缀
        executor.initialize();
        return executor;
    }
}
3.2 定义异步服务层

创建一个服务类，其中包含实际执行耗时逻辑的方法。该方法需要用@Async注解标记，并指定使用的TaskExecutor。为了支持取消，该方法应返回一个CompletableFuture，并且在内部逻辑中周期性地检查线程中断状态。

Post AI Post AI

博客文章AI生成器

Post AI50 查看详情 Post AI 
// src/main/java/com/example/async/LongRunningService.java
package com.example.async;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class LongRunningService {

    private static final Logger logger = LoggerFactory.getLogger(LongRunningService.class);

    @Async("taskExecutor") // 指定使用名为"taskExecutor"的线程池
    public CompletableFuture<String> executeLongRunningTask(String taskId, int timeToRun) {
        logger.info("Task {} started with timeToRun: {}", taskId, timeToRun);
        try {
            for (int i = 0; i < timeToRun; i++) {
                // 模拟耗时操作
                Thread.sleep(1000); // 每次操作耗时1秒

                // 检查线程是否被中断
                if (Thread.currentThread().isInterrupted()) {
                    logger.warn("Task {} was interrupted after {} seconds.", taskId, i + 1);
                    return CompletableFuture.completedFuture("Task " + taskId + " cancelled.");
                }
                logger.info("Task {} processing... {}/{} seconds", taskId, i + 1, timeToRun);
            }
            logger.info("Task {} completed successfully.", taskId);
            return CompletableFuture.completedFuture("Task " + taskId + " completed.");
        } catch (InterruptedException e) {
            // 捕获InterruptedException，通常表示线程被中断
            Thread.currentThread().interrupt(); // 重新设置中断标志
            logger.warn("Task {} caught InterruptedException. It was cancelled.", taskId);
            return CompletableFuture.completedFuture("Task " + taskId + " cancelled due to interruption.");
        } catch (Exception e) {
            logger.error("Task {} encountered an error: {}", taskId, e.getMessage());
            return CompletableFuture.completedFuture("Task " + taskId + " failed: " + e.getMessage());
        }
    }
}

关键点：

  • @Async("taskExecutor")：确保此方法在自定义的线程池中运行。
  • CompletableFuture<String>：允许我们异步地获取任务结果，并且提供了取消的能力。
  • Thread.currentThread().isInterrupted()：在耗时循环中周期性检查中断状态。这是实现任务取消的关键。当Future.cancel(true)被调用时，它会尝试中断底层线程，此时isInterrupted()会返回true。
  • InterruptedException：当Thread.sleep()等阻塞方法在线程被中断时抛出。捕获它并重新设置中断标志是良好实践。
3.3 构建API控制器与任务管理

我们需要一个控制器来接收启动任务和取消任务的请求。为了管理任务，我们将使用一个Map来存储任务ID和对应的CompletableFuture。

// src/main/java/com/example/async/TaskController.java
package com.example.async;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CompletableFuture;

@RestController
@RequestMapping("/api/tasks")
public class TaskController {

    private static final Logger logger = LoggerFactory.getLogger(TaskController.class);
    private final LongRunningService longRunningService;

    // 使用ConcurrentHashMap来安全地存储任务Future
    private final Map<String, CompletableFuture<String>> runningTasks = new ConcurrentHashMap<>();

    public TaskController(LongRunningService longRunningService) {
        this.longRunningService = longRunningService;
    }

    /**
     * 启动一个长耗时任务
     * @param timeToRun 模拟任务运行时间（秒）
     * @return 包含任务ID的响应
     */
    @PostMapping("/start/{timeToRun}")
    public ResponseEntity<String> startTask(@PathVariable int timeToRun) {
        String taskId = UUID.randomUUID().toString();
        logger.info("Received request to start task {} with timeToRun: {}", taskId, timeToRun);

        CompletableFuture<String> future = longRunningService.executeLongRunningTask(taskId, timeToRun);
        runningTasks.put(taskId, future);

        // 异步监听任务完成，并从map中移除
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                logger.info("Task {} completed with result: {}", taskId, result);
            } else {
                logger.error("Task {} failed with exception: {}", taskId, ex.getMessage());
            }
            runningTasks.remove(taskId); // 任务完成后从map中移除
        });

        return ResponseEntity.ok("Task started. Task ID: " + taskId);
    }

    /**
     * 取消一个正在运行的任务
     * @param taskId 要取消的任务ID
     * @return 取消结果
     */
    @PostMapping("/cancel/{taskId}")
    public ResponseEntity<String> cancelTask(@PathVariable String taskId) {
        CompletableFuture<String> future = runningTasks.get(taskId);

        if (future == null) {
            logger.warn("Attempted to cancel non-existent or already completed task: {}", taskId);
            return ResponseEntity.badRequest().body("Task " + taskId + " not found or already completed.");
        }

        if (future.isDone()) {
            logger.info("Task {} is already done, cannot cancel.", taskId);
            runningTasks.remove(taskId); // 确保已完成的任务被移除
            return ResponseEntity.badRequest().body("Task " + taskId + " is already done, cannot cancel.");
        }

        // 尝试取消任务，true表示如果任务正在运行，会尝试中断线程
        boolean cancelled = future.cancel(true);
        if (cancelled) {
            logger.info("Task {} successfully requested for cancellation.", taskId);
            // 任务被成功取消后，也从map中移除
            runningTasks.remove(taskId);
            return ResponseEntity.ok("Task " + taskId + " cancellation requested.");
        } else {
            logger.warn("Task {} could not be cancelled. It might be completed or not interruptible.", taskId);
            return ResponseEntity.status(500).body("Failed to cancel task " + taskId + ".");
        }
    }

    /**
     * 获取所有正在运行的任务ID
     * @return 正在运行的任务ID列表
     */
    @GetMapping("/status")
    public ResponseEntity<Map<String, String>> getRunningTasks() {
        Map<String, String> statusMap = new ConcurrentHashMap<>();
        runningTasks.forEach((id, future) -> {
            statusMap.put(id, future.isDone() ? "DONE" : (future.isCancelled() ? "CANCELLED" : "RUNNING"));
        });
        return ResponseEntity.ok(statusMap);
    }
}
4. 运行与测试
  1. 启动Spring Boot应用。
  2. 启动任务：
    • 使用POST请求访问 http://localhost:8080/api/tasks/start/10 (启动一个运行10秒的任务)
    • 响应会返回一个任务ID，例如 Task started. Task ID: 123e4567-e89b-12d3-a456-426614174000
  3. 查询任务状态：
    • 使用GET请求访问 http://localhost:8080/api/tasks/status
    • 可以看到正在运行的任务及其状态。
  4. 取消任务：
    • 在任务运行期间，使用POST请求访问 http://localhost:8080/api/tasks/cancel/{taskId}，将 {taskId} 替换为实际的任务ID。
    • 观察控制台日志，会看到服务层打印出任务被中断的信息。
5. 注意事项与最佳实践
  • 线程池配置： ThreadPoolTaskExecutor的corePoolSize、maxPoolSize和queueCapacity需要根据应用负载和服务器资源进行合理配置，以避免线程饥饿或资源耗尽。
  • 优雅中断： 任务内部必须周期性地检查Thread.currentThread().isInterrupted()，并根据其状态决定是否退出。如果任务中包含第三方库调用，这些库可能不会响应中断，需要额外处理或避免在其中进行中断。
  • 资源清理： runningTasks Map必须妥善管理。在任务完成（无论成功、失败或取消）后，应及时从Map中移除对应的Future，防止内存泄漏。CompletableFuture.whenComplete()方法提供了一个方便的清理机制。
  • 错误处理： 异步任务中的异常应该被捕获并妥善处理。CompletableFuture提供了exceptionally()和handle()等方法来处理异步操作中的异常。
  • 用户反馈： 客户端可能需要知道任务的实时状态。可以考虑使用WebSocket或Server-Sent Events (SSE) 来向客户端推送任务状态更新。
  • 持久化任务： 对于需要确保执行且不能丢失的任务，仅靠内存中的Map是不够的。应考虑将任务信息持久化到数据库，并结合消息队列（如Kafka, RabbitMQ）或专门的作业调度框架（如Quartz, Activiti）来实现更健壮的任务管理和恢复机制。
  • 安全性： 确保只有授权用户才能启动或取消任务。在API层增加认证和授权机制。
  • 超时机制： 除了手动取消，还可以为任务设置超时时间。CompletableFuture结合ScheduledExecutorService可以实现超时取消。
6. 总结

通过将长耗时操作异步化并结合任务管理机制，我们能够在Spring Boot中实现对API请求的动态取消。这种模式不仅提升了API的响应性和用户体验，也增强了系统的资源利用效率和稳定性。理解异步编程的核心概念、正确配置线程池以及在任务内部响应中断是实现这一功能的关键。在实际项目中，还需要根据具体需求，进一步考虑任务持久化、实时状态反馈和安全性等高级特性。

以上就是在Spring Boot中实现长耗时API请求的异步执行与取消策略的详细内容，更多请关注知识资源分享宝库其它相关文章！

相关标签： java app websocket ai 并发请求 red Java spring rabbitmq spring boot kafka String for 标识符 循环 接口 线程 Thread map 并发 对象 异步 数据库 http websocket 大家都在看： Java中序列化和反序列化的核心原理 Java中Calendar类常用方法解析 Java字节码中匿名内部类的命名约定与原理 Java中Octet类加法操作的实现与二进制处理 Java中自定义8位二进制数类Octet的加法实现教程

标签:  耗时 请求 取消 

相关阅读

发表评论:

◎欢迎参与讨论，请在这里发表您的看法、交流您的观点。

最近发表

标签列表