Springboot中的異步任務(wù)執(zhí)行及監(jiān)控詳解

更新時(shí)間：2023年10月19日 09:01:07 作者：shirukai

這篇文章主要介紹了Springboot中的異步任務(wù)執(zhí)行及監(jiān)控詳解,除了自己實(shí)現(xiàn)線(xiàn)程外,springboot本身就提供了通過(guò)注解的方式,進(jìn)行異步任務(wù)的執(zhí)行,下面主要記錄一下,在Springboot項(xiàng)目中實(shí)現(xiàn)異步任務(wù),以及對(duì)異步任務(wù)進(jìn)行封裝監(jiān)控,需要的朋友可以參考下

前言

除了自己實(shí)現(xiàn)線(xiàn)程外，springboot本身就提供了通過(guò)注解的方式，進(jìn)行異步任務(wù)的執(zhí)行。

下面主要記錄一下，在Springboot項(xiàng)目中實(shí)現(xiàn)異步任務(wù)，以及對(duì)異步任務(wù)進(jìn)行封裝監(jiān)控。

1 開(kāi)啟異步支持

想要使用springboot的注解進(jìn)行異步任務(wù)，首先要開(kāi)啟springboot的異步任務(wù)支持。通過(guò)集成AsyncConfigurer接口，并實(shí)現(xiàn)getAsyncExcutor()方法，如下所示：

package com.springboot.demo.asyncTask.conf;

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;

/**
 * Created by shirukai on 2018/7/30
 * 配置spring boot 多線(xiàn)程支持
 */
@Configuration
@EnableAsync   //開(kāi)啟異步任務(wù)支持
public class SpringTaskExecutor implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(5);
        taskExecutor.setMaxPoolSize(10);
        taskExecutor.setQueueCapacity(20);
        taskExecutor.initialize();
        return taskExecutor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return null;
    }
}

通過(guò)上面方法，就可以實(shí)現(xiàn)Spring boot的異步任務(wù)支持。然后只需要在想要進(jìn)行異步的方法前添加@Async注解就可以了，如下圖所示：

package com.springboot.demo.asyncTask.service;

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

/**
 * Created by shirukai on 2018/7/31
 * 異步任務(wù)執(zhí)行器
 */
@Component
public class AsyncTaskExecutor {
    private static Logger LOG = LoggerFactory.getLogger(AsyncTaskExecutor.class);
    @Async
    public void executor(AsyncTaskConstructor asyncTaskGenerator, String taskInfo) {
        LOG.info("AsyncTaskExecutor is executing async task:{}", taskInfo);
        asyncTaskGenerator.async();
    }
}

2 異步任務(wù)封裝監(jiān)控

2.1 封裝思路

提供一個(gè)異步任務(wù)的管理器，管理器可以實(shí)現(xiàn)異步任務(wù)的提交、保存任務(wù)信息、獲取任務(wù)信息等功能。

提供一個(gè)異步任務(wù)的監(jiān)控器，用于監(jiān)控異步任務(wù)執(zhí)行狀況，并把執(zhí)行信息保存到緩存中，并記錄任務(wù)執(zhí)行時(shí)間。

提供一個(gè)異步任務(wù)的構(gòu)造器，用于構(gòu)造異步方法。

提供一個(gè)異步任務(wù)的執(zhí)行器，用于執(zhí)行管理器提交的使用構(gòu)造器構(gòu)造的異步方法。

2.2 效果展示

2.2.1 啟動(dòng)異步任務(wù)

2.2.2 查看任務(wù)狀態(tài)

2.3 編碼實(shí)現(xiàn)

2.3.1 conf包

主要是配置springboot的線(xiàn)程池，開(kāi)啟spring boot支持異步支持

SpringTaskExcutor.java

package com.emcc.hiacloud.analytics.asynctask.conf;

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;

/**
 * Created by shirukai on 2018/7/30
 * 配置spring boot 多線(xiàn)程支持
 */
@Configuration
@EnableAsync   //開(kāi)啟異步任務(wù)支持
public class SpringTaskExecutor implements AsyncConfigurer {
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(5);
        taskExecutor.setMaxPoolSize(10);
        taskExecutor.setQueueCapacity(20);
        taskExecutor.initialize();
        return taskExecutor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return null;
    }
}

2.3.2 entity包

主要存放TaskInfo實(shí)例類(lèi)用于保存任務(wù)信息，TaskStatusEnmu枚舉類(lèi)用來(lái)存放任務(wù)狀態(tài)。

TaskInfo.java

package com.emcc.hiacloud.analytics.asynctask.entity;


import java.util.Date;

/**
 * Created by shirukai on 2018/7/31
 * 任務(wù)信息
 */
public class TaskInfo {
    private String taskId;
    private TaskStatusEnum status;
    private Date startTime;
    private Date endTime;
    private String totalTime;

    public TaskStatusEnum getStatus() {
        return status;
    }

    public void setStatus(TaskStatusEnum status) {
        this.status = status;
    }

    public void setTotalTime(String totalTime) {
        this.totalTime = totalTime;
    }

    public String getTaskId() {
        return taskId;
    }

    public void setTaskId(String taskId) {
        this.taskId = taskId;
    }

    public Date getStartTime() {
        return startTime;
    }

    public void setStartTime(Date startTime) {
        this.startTime = startTime;
    }

    public Date getEndTime() {
        return endTime;
    }

    public void setEndTime(Date endTime) {
        this.endTime = endTime;
    }

    public String getTotalTime() {
        return totalTime;
    }

    public void setTotalTime() {
        this.totalTime = (this.endTime.getTime() - this.startTime.getTime()) + "ms";
    }
}

TaskStatusEnum.java

package com.emcc.hiacloud.analytics.asynctask.entity;

/**
 * Created by shirukai on 2018/7/31
 * 任務(wù)狀態(tài)枚舉
 */
public enum TaskStatusEnum {

    STARTED(1, "任務(wù)已經(jīng)啟動(dòng)"),
    RUNNING(0, "任務(wù)正在運(yùn)行"),
    SUCCESS(2, "任務(wù)執(zhí)行成功"),
    FAILED(-2, "任務(wù)執(zhí)行失敗");
    private int state;
    private String stateInfo;

    TaskStatusEnum(int state, String stateInfo) {
        this.state = state;
        this.stateInfo = stateInfo;
    }

    public int getState() {
        return state;
    }

    public String getStateInfo() {
        return stateInfo;
    }

}

2.3.3 manager包

存放要不任務(wù)的管理類(lèi)和監(jiān)控類(lèi)

AsyncTaskManager.java

package com.emcc.hiacloud.analytics.asynctask.manager;


import com.emcc.hiacloud.analytics.asynctask.entity.TaskInfo;
import com.emcc.hiacloud.analytics.asynctask.entity.TaskStatusEnum;
import com.emcc.hiacloud.analytics.asynctask.service.AsyncTaskConstructor;
import com.emcc.hiacloud.analytics.asynctask.service.AsyncTaskExecutor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * Created by shirukai on 2018/7/31
 * 異步任務(wù)管理器
 */
@Component
public class AsyncTaskManager {
    private Map<String, TaskInfo> taskContainer = new HashMap<>(16);
    @Autowired
    AsyncTaskExecutor asyncTaskExecutor;


    /**
     * 初始化任務(wù)
     *
     * @return taskInfo
     */
    public TaskInfo initTask() {
        TaskInfo taskInfo = new TaskInfo();
        taskInfo.setTaskId(getTaskId());
        taskInfo.setStatus(TaskStatusEnum.STARTED);
        taskInfo.setStartTime(new Date());
        setTaskInfo(taskInfo);
        return taskInfo;
    }

    /**
     * 初始化任務(wù)
     * @param asyncTaskConstructor 異步任務(wù)構(gòu)造器
     * @return taskInfo
     */
    public TaskInfo submit(AsyncTaskConstructor asyncTaskConstructor) {
        TaskInfo info = initTask();
        String taskId = info.getTaskId();
        asyncTaskExecutor.executor(asyncTaskConstructor,taskId);
        return info;
    }

    /**
     * 保存任務(wù)信息
     *
     * @param taskInfo 任務(wù)信息
     */
    public void setTaskInfo(TaskInfo taskInfo) {
        taskContainer.put(taskInfo.getTaskId(), taskInfo);
    }

    /**
     * 獲取任務(wù)信息
     *
     * @param taskId 任務(wù)ID
     * @return
     */
    public TaskInfo getTaskInfo(String taskId) {
        return taskContainer.get(taskId);
    }

    /**
     * 獲取任務(wù)狀態(tài)
     *
     * @param taskId 任務(wù)ID
     * @return
     */
    public TaskStatusEnum getTaskStatus(String taskId) {
        return getTaskInfo(taskId).getStatus();
    }

    /**
     * 生成任務(wù)ID
     *
     * @return taskId
     */
    public String getTaskId() {
        return UUID.randomUUID().toString();
    }
}

AsyncTaskMonitor.java

異步任務(wù)的監(jiān)控主要是利用了spring的AOP面向切面，在異步方法的執(zhí)行前和執(zhí)行后進(jìn)行監(jiān)控，判斷任務(wù)狀態(tài)，并記錄任務(wù)信息。

package com.emcc.hiacloud.analytics.asynctask.manager;


import com.emcc.hiacloud.analytics.asynctask.entity.TaskInfo;
import com.emcc.hiacloud.analytics.asynctask.entity.TaskStatusEnum;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * Created by shirukai on 2018/7/31
 * 異步任務(wù)監(jiān)控
 */
@Component
@Aspect
public class AsyncTaskMonitor {
    @Autowired
    AsyncTaskManager manager;
    private static Logger LOG = LoggerFactory.getLogger(AsyncTaskMonitor.class);

    @Around("execution(* com.emcc.hiacloud.analytics.asynctask.service.AsyncTaskExecutor.*(..))")
    public void taskHandle(ProceedingJoinPoint pjp) {
        //獲取taskId
        String taskId = pjp.getArgs()[1].toString();
        //獲取任務(wù)信息
        TaskInfo taskInfo = manager.getTaskInfo(taskId);
        LOG.info("AsyncTaskMonitor is monitoring async task:{}", taskId);
        taskInfo.setStatus(TaskStatusEnum.RUNNING);
        manager.setTaskInfo(taskInfo);
        TaskStatusEnum status = null;
        try {
            pjp.proceed();
            status = TaskStatusEnum.SUCCESS;
        } catch (Throwable throwable) {
            status = TaskStatusEnum.FAILED;
            LOG.error("AsyncTaskMonitor:async task {} is failed.Error info:{}", taskId, throwable.getMessage());
        }
        taskInfo.setEndTime(new Date());
        taskInfo.setStatus(status);
        taskInfo.setTotalTime();
        manager.setTaskInfo(taskInfo);
    }
}

2.3.4 service包

主要存放異步任務(wù)的方法構(gòu)造器和執(zhí)行器。

AsyncTaskConstructor

通過(guò)該接口可以構(gòu)建想要實(shí)現(xiàn)的異步方法。只要new 一下接口實(shí)例，然后重寫(xiě)李曼的async()方法即可。

package com.emcc.hiacloud.analytics.asynctask.service;

/**
 * Created by shirukai on 2018/7/31
 * 異步任務(wù)構(gòu)造器
 */
public interface AsyncTaskConstructor {
    public void async();
}

AsyncTaskExecutor.java

異步任務(wù)執(zhí)行器

package com.emcc.hiacloud.analytics.asynctask.service;

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

/**
 * Created by shirukai on 2018/7/31
 * 異步任務(wù)執(zhí)行器
 */
@Component
public class AsyncTaskExecutor {
    private static Logger LOG = LoggerFactory.getLogger(AsyncTaskExecutor.class);
    @Async
    public void executor(AsyncTaskConstructor asyncTaskGenerator, String taskInfo) {
        LOG.info("AsyncTaskExecutor is executing async task:{}", taskInfo);
        asyncTaskGenerator.async();
    }
}

3 應(yīng)用

實(shí)現(xiàn)兩個(gè)接口，一個(gè)是開(kāi)啟一個(gè)異步任務(wù)，另一個(gè)是查看任務(wù)狀態(tài)。

想要使用我們剛才分裝好的異步任務(wù)，只需要將AsyncTaskManager注入到程序中。

package com.emcc.hiacloud.analytics.orchestrations.controller;

import com.emcc.hiacloud.analytics.asynctask.entity.TaskInfo;
import com.emcc.hiacloud.analytics.asynctask.manager.AsyncTaskManager;
import com.emcc.hiacloud.analytics.common.rest.RestMessage;
import com.emcc.hiacloud.analytics.common.util.RestMessageUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * Created by shirukai on 2018/7/31
 */
@RestController
@RequestMapping(value = "/api/v1/asynctask")
public class AsyncTaskController {
    //注入異步任務(wù)管理器
    @Autowired
    AsyncTaskManager asyncTaskManager;

    @RequestMapping(value = "/startTask", method = RequestMethod.GET)
    public RestMessage startAsyncTask() {
        //調(diào)用任務(wù)管理器中的submit去提交一個(gè)異步任務(wù)
        TaskInfo taskInfo = asyncTaskManager.submit(() -> {
            System.out.println("__________");
            try {
                //模擬異步，睡眠6秒
                Thread.sleep(30000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("__________");
        });
        return RestMessageUtil.objectToRestMessage(taskInfo);
    }

    @RequestMapping(value = "/getTaskStatus", method = RequestMethod.GET)
    public RestMessage getTaskStatus(
            @RequestParam("taskId") String taskId) {
        return RestMessageUtil.objectToRestMessage(asyncTaskManager.getTaskInfo(taskId));
    }
}

到此這篇關(guān)于Springboot中的異步任務(wù)執(zhí)行及監(jiān)控詳解的文章就介紹到這了,更多相關(guān)Springboot異步任務(wù)及監(jiān)控內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: