springboot如何添加task任務(wù)執(zhí)行隊列

更新時間：2023年07月21日 14:58:57 作者：huayang183

這篇文章主要介紹了springboot如何添加task任務(wù)執(zhí)行隊列問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

springboot添加task任務(wù)執(zhí)行隊列

主要分為三塊：任務(wù)task，TaskExecutor任務(wù)執(zhí)行器，TaskQueue任務(wù)隊列。

下面詳細講解一下：

任務(wù)task的接口ITask

public interface ITask {
? ?//執(zhí)行方法
? ? void run();
}

任務(wù)task的接口實現(xiàn)類

public class ActionAppTask implements ITask {
? ? public ActionAppTask(){
? ? ? //構(gòu)造方法
? ? }
? ? @Override
? ? public void run() {
? ? ? //todo:要執(zhí)行的操作
? ? }
}

TaskExecutor任務(wù)執(zhí)行器

public class TaskExecutor extends Thread {
? ? // 任務(wù)隊列，里面是要執(zhí)行的任務(wù)。
? ? private BlockingQueue<ITask> taskQueue;
? ? // 任務(wù)隊列是否在執(zhí)行任務(wù)
? ? private boolean isRunning = true;
? ? public TaskExecutor(BlockingQueue<ITask> taskQueue) {
? ? ? ? this.taskQueue = taskQueue;
? ? }
? ? // 退出。
? ? public void quit() {
? ? ? ? isRunning = false;
? ? ? ? interrupt();
? ? }
? ? @Override
? ? public void run() {
? ? ? ? while (isRunning) { // 如果是執(zhí)行狀態(tài)就待著。
? ? ? ? ? ? ITask iTask;
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? iTask = taskQueue.take(); // 下一個任務(wù)，沒有就等著。
? ? ? ? ? ? } catch (InterruptedException e) {
? ? ? ? ? ? ? ? if (!isRunning) {
? ? ? ? ? ? ? ? ? ? // 發(fā)生意外了，是退出狀態(tài)的話就把窗口關(guān)閉。
? ? ? ? ? ? ? ? ? ? interrupt();
? ? ? ? ? ? ? ? ? ? break; // 如果執(zhí)行到break，后面的代碼就無效了。
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? // 發(fā)生意外了，不是退出狀態(tài)，那么窗口繼續(xù)等待。
? ? ? ? ? ? ? ? continue;
? ? ? ? ? ? }
? ? ? ? ? ? // 執(zhí)行任務(wù)。
? ? ? ? ? ? iTask.run();
? ? ? ? }
? ? }
}

TaskQueue任務(wù)隊列

public class TaskQueue {
? ? // 隊列，里面是任務(wù)。
? ? private BlockingQueue<ITask> mTaskQueue;
? ? // 好多執(zhí)行器。
? ? private TaskExecutor[] mTaskExecutors;
? ? // new隊列的時候，要指定執(zhí)行器數(shù)量。這可以確定你開的多個線程是否需要等待。
? ? public TaskQueue(int size) {
? ? ? ? mTaskQueue = new LinkedBlockingQueue<>();
? ? ? ? mTaskExecutors = new TaskExecutor[size];
? ? }
? ? // 開啟隊列。
? ? public void start() {
? ? ? ? stop();
? ? ? ? // 開啟隊列。
? ? ? ? for (int i = 0; i < mTaskExecutors.length; i++) {
? ? ? ? ? ? mTaskExecutors[i] = new TaskExecutor(mTaskQueue);
? ? ? ? ? ? mTaskExecutors[i].start();
? ? ? ? }
? ? }
? ? // 關(guān)閉隊列。
? ? public void stop() {
? ? ? ? if (mTaskExecutors != null)
? ? ? ? ? ? for (TaskExecutor taskExecutor : mTaskExecutors) {
? ? ? ? ? ? ? ? if (taskExecutor != null) taskExecutor.quit();
? ? ? ? ? ? }
? ? }
? ? //添加任務(wù)到隊列。
? ? public <T extends ITask> int add(T task) {
? ? ? ? if (!mTaskQueue.contains(task)) {
? ? ? ? ? ? mTaskQueue.add(task);
? ? ? ? }
? ? ? ? // 返回隊列中的任務(wù)數(shù)
? ? ? ? return mTaskQueue.size();
? ? }
}

親測非常好用，對于有線程返回值的異步任務(wù)處理，非常適合。

springboot添加task異步任務(wù)隊列

public interface Itask {
    void run();
}

public class MyTask implements Itask {
    public MyTask () {
    }
	//要執(zhí)行的代碼邏輯
    @Override
    public void run() {
        System.out.println("MyTask");
    }
}

public class TaskExecutor extends Thread {
    private BlockingQueue taskQueue;
    private Boolean isRunning = true;
    public TaskExecutor(BlockingQueue taskQueue) {
        this.taskQueue = taskQueue;
    }
    // 退出。
    public void quit() {
        isRunning = false;
        interrupt();
    }
    @Override
    public void run() {
        while (isRunning) { // 如果是執(zhí)行狀態(tài)就待著。
            Itask iTask;
            try {
                iTask = (Itask) taskQueue.take(); // 下一個任務(wù)，沒有就等著。
            } catch (InterruptedException e) {
                if (!isRunning) {
                    // 發(fā)生錯誤中斷代碼
                    interrupt();
                    break; 
                }
                // 發(fā)生意外了，不是退出狀態(tài)，那么窗口繼續(xù)等待。
                continue;
            }
            // 執(zhí)行任務(wù)
            iTask.run();
        }
    }
}

public class TaskQueue {
    // 任務(wù)隊列
    private BlockingQueue mTaskQueue;
    // 執(zhí)行器
    private TaskExecutor[] mTaskExecutors;
    // 創(chuàng)建隊列時,指定執(zhí)行器數(shù)量,保證你開的多個線程是否需要等待
    public TaskQueue(int size) {
        mTaskQueue = new LinkedBlockingQueue<>();
        mTaskExecutors = new TaskExecutor[size];
    }
    // 開啟隊列。
    public void start() {
        stop();
        for (int i = 0; i < mTaskExecutors.length; i++) {
            mTaskExecutors[i] = new TaskExecutor(mTaskQueue);
            mTaskExecutors[i].start();
        }
    }
    // 關(guān)閉隊列。
    public void stop() {
        if (mTaskExecutors != null)
            for (TaskExecutor taskExecutor : mTaskExecutors) {
                if (taskExecutor != null) taskExecutor.quit();
            }
    }
    //添加任務(wù)到隊列。
    public  int add(Itask task) {
        if (!mTaskQueue.contains(task)) {
            mTaskQueue.add(task);
        }
        // 返回隊列中的任務(wù)數(shù)
        return mTaskQueue.size();
    }
}

public class Test{
	public static void main(String[] args) {
        TaskQueue tq = new TaskQueue(1);
        MyTask myTask= new MyTask ();
        tq.add(myTask);
        tq.start();
    }
}