Spring Boot：如何在运行时暂停@Async注解方法执行并保留队列？

当然有办法实现这个需求！本质上我们需要给异步任务的执行加上一个"闸门"——在CSV文件解析期间关闭闸门，让异步任务先进入队列等待；解析完成后打开闸门，让队列里的任务依次执行。下面给你两个实用的方案：

方案一：自定义可控的Async执行器 #

我们可以自定义一个ThreadPoolTaskExecutor，给它加一个全局的控制开关，结合Spring的@Async注解使用。

步骤1：创建带开关的执行器 #

首先定义一个自定义的执行器类，里面包含开关变量和任务队列控制逻辑：

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class ControllableAsyncExecutor extends ThreadPoolTaskExecutor {
    // 控制开关，默认开启任务执行
    private volatile boolean isTaskAllowed = true;

    @Override
    protected void initializeExecutor(ThreadPoolExecutor executor, ThreadPoolExecutor.CallerRunsPolicy rejectedExecutionHandler) {
        super.initializeExecutor(executor, new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                // 如果任务不允许执行，直接加入队列等待（可根据业务调整队列策略）
                if (!isTaskAllowed) {
                    try {
                        executor.getQueue().put(r);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                } else {
                    rejectedExecutionHandler.rejectedExecution(r, executor);
                }
            }
        });
    }

    // 暂停异步任务执行
    public void pauseTasks() {
        this.isTaskAllowed = false;
    }

    // 恢复异步任务执行，唤醒队列等待的任务
    public void resumeTasks() {
        this.isTaskAllowed = true;
        BlockingQueue<Runnable> queue = getThreadPoolExecutor().getQueue();
        synchronized (queue) {
            queue.notifyAll();
        }
    }

    public boolean isTaskAllowed() {
        return isTaskAllowed;
    }
}

步骤2：配置Spring使用这个执行器 #

在配置类里把自定义执行器注册为Spring的异步执行器：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean("controllableExecutor")
    public ControllableAsyncExecutor controllableAsyncExecutor() {
        ControllableAsyncExecutor executor = new ControllableAsyncExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(100); // 根据你的任务量调整队列大小
        executor.setThreadNamePrefix("ControllableAsync-");
        executor.initialize();
        return executor;
    }
}

步骤3：在@Async方法上指定执行器 #

在你的异步委托类中，指定使用这个自定义执行器：

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

@Component
public class AsyncDelegate {

    @Async("controllableExecutor")
    public void executeResourceIntensiveTask() {
        // 你的资源密集型任务逻辑
    }
}

步骤4：在CSV解析任务中控制开关 #

假设你用定时任务触发CSV解析，在解析前后调用执行器的开关：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class CsvProcessingTask {

    @Autowired
    private ControllableAsyncExecutor controllableAsyncExecutor;

    // 每天凌晨2点执行CSV解析（可根据需求调整cron表达式）
    @Scheduled(cron = "0 0 2 * * ?")
    public void processCsvFile() {
        // 1. 暂停异步任务执行
        controllableAsyncExecutor.pauseTasks();
        try {
            // 2. 执行CSV文件读取和解析逻辑
            readAndParseLargeCsv();
        } finally {
            // 3. 解析完成后恢复异步任务执行
            controllableAsyncExecutor.resumeTasks();
        }
    }

    private void readAndParseLargeCsv() {
        // 你的CSV解析代码
    }
}

方案二：使用信号量实现解耦控制 #

另一种更解耦的方式是通过全局信号量，让异步方法在执行前等待允许执行的信号。

步骤1：定义全局控制信号量 #

创建一个全局Bean来管理任务执行的信号：

import org.springframework.stereotype.Component;
import java.util.concurrent.CountDownLatch;

@Component
public class TaskControlSignal {
    // 初始值为1，表示允许任务执行；暂停时重置为0让任务等待
    private volatile CountDownLatch latch = new CountDownLatch(1);

    // 暂停任务：重置信号量，让后续任务进入等待状态
    public void pause() {
        this.latch = new CountDownLatch(1);
    }

    // 恢复任务：释放信号量，唤醒所有等待的任务
    public void resume() {
        this.latch.countDown();
    }

    // 等待允许执行的信号
    public void await() throws InterruptedException {
        latch.await();
    }
}

步骤2：在异步方法中加入等待逻辑 #

在@Async方法执行前先等待信号量：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

@Component
public class AsyncDelegate {

    @Autowired
    private TaskControlSignal taskControlSignal;

    @Async
    public void executeResourceIntensiveTask() throws InterruptedException {
        // 等待允许执行的信号
        taskControlSignal.await();
        // 你的资源密集型任务逻辑
    }
}

步骤3：在CSV解析任务中控制信号量 #

同样在CSV解析前后调用信号量的暂停和恢复：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class CsvProcessingTask {

    @Autowired
    private TaskControlSignal taskControlSignal;

    @Scheduled(cron = "0 0 2 * * ?")
    public void processCsvFile() {
        // 暂停任务
        taskControlSignal.pause();
        try {
            readAndParseLargeCsv();
        } finally {
            // 恢复任务
            taskControlSignal.resume();
        }
    }

    private void readAndParseLargeCsv() {
        // CSV解析逻辑
    }
}

注意事项 #

• 如果你使用无界队列，要注意内存溢出风险，建议设置合理的队列容量并搭配合适的拒绝策略。
• 所有控制开关变量都要用volatile修饰，保证多线程下的可见性。
• 如果存在多个CSV解析任务并发触发的场景，建议给信号量控制逻辑加锁，确保原子性。

内容的提问来源于stack exchange，提问作者user3479176