package com.example.demo.dto;

import java.io.Serializable;

public class Type implements Serializable {

    private String code;

    private String name;

    public Type(String code, String name) {

        this.code = code;

        this.name = name;

    }

    public String getCode() {

        return code;

    }

    public void setCode(String code) {

        this.code = code;

    }

    public String getName() {

        return name;

    }

    public void setName(String name) {

        this.name = name;

    }

}

package com.example.demo.entity;

import com.example.demo.dto.Type;

import net.minidev.json.annotate.JsonIgnore;

import org.springframework.data.annotation.Id;

import org.springframework.data.redis.core.RedisHash;

import org.springframework.data.redis.core.TimeToLive;

import java.io.Serializable;

import java.util.concurrent.TimeUnit;

@RedisHash(value = "demo")

public class ProgressTaskCache implements Serializable {

    @Id

    private String id;

    /**

     * 任务状态

     */

    private Type status;

    /**

     * 任务消息

     */

    private String message;

    /**

     * 结果数据

     */

    protected String data;

    /**

     * 存活时间。默认 10 分钟

     */

    @TimeToLive(unit = TimeUnit.MILLISECONDS)

    @JsonIgnore

    private long timeout = 600000;

    public long getTimeout() {

        return timeout;

    }

    public void setTimeout(long timeout) {

        this.timeout = timeout;

    }

    public String getId() {

        return id;

    }

    public void setId(String id) {

        this.id = id;

    }

    public Type getStatus() {

        return status;

    }

    public void setStatus(Type status) {

        this.status = status;

    }

    public String getMessage() {

        return message;

    }

    public void setMessage(String message) {

        this.message = message;

    }

    public String getData() {

        return data;

    }

    public void setData(String data) {

        this.data = data;

    }

}

package com.example.demo.dto;

import com.example.demo.entity.ProgressTaskCache;

import java.io.Serializable;

public class ResponseResult implements Serializable {

    /**

     * 代码.

     */

    protected int code;

    /**

     * 消息.

     */

    protected String msg;

    /**

     * 状态.

     */

    protected boolean success;

    /**

     * 异步任务.

     */

    private ProgressTaskCache progressTask;

    public int getCode() {

        return code;

    }

    public void setCode(int code) {

        this.code = code;

    }

    public String getMsg() {

        return msg;

    }

    public void setMsg(String msg) {

        this.msg = msg;

    }

    public boolean isSuccess() {

        return success;

    }

    public void setSuccess(boolean success) {

        this.success = success;

    }

    public ProgressTaskCache getProgressTask() {

        return progressTask;

    }

    public void setProgressTask(ProgressTaskCache progressTask) {

        this.progressTask = progressTask;

    }

}

package com.example.demo.util;

import com.fasterxml.jackson.core.JsonProcessingException;

import com.fasterxml.jackson.databind.DeserializationFeature;

import com.fasterxml.jackson.databind.ObjectMapper;

import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import javax.validation.ValidationException;

import java.io.IOException;

public final class ConvertUtil {

    private static final Logger LOG = LoggerFactory.getLogger(ConvertUtil.class);

    private static ObjectMapper objectMapper;

    public static ObjectMapper getObjectMapper() {

        if (objectMapper == null) {

            objectMapper = new ObjectMapper();

            objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

            objectMapper.registerModule(new JavaTimeModule());

        }

        return objectMapper;

    }

    /**

     * 对象转字符串.

     *

     * @param object 对象.

     * @return json 字符串.

     */

    public static String toJson(Object object) {

        String json = null;

        try {

            json = getObjectMapper().writeValueAsString(object);

        } catch (JsonProcessingException e) {

            e.printStackTrace();

            LOG.error("序列化失败：" , e);

            throw new ValidationException("序列化失败");

        }

        return json;

    }

    /**

     * 从 json 字符串转成对象.

     *

     * @param json  json 字符串.

     * @param clazz 类.

     * @param <T>   枚举值.

     * @return 对象.

     */

    public static <T> T fromJson(String json, Class<T> clazz) {

        T t = null;

        try {

            t = getObjectMapper().readValue(json, clazz);

        } catch (IOException e) {

            e.printStackTrace();

        }

        return t;

    }

}

package com.example.demo.dao;

import com.example.demo.entity.ProgressTaskCache;

import org.springframework.data.keyvalue.repository.KeyValueRepository;

import org.springframework.stereotype.Repository;

@Repository

public interface ProgressTaskCacheDao extends KeyValueRepository<ProgressTaskCache, String> {

    boolean expireAt(String id, long delay);

}

package com.example.demo.dao;

import com.example.demo.entity.ProgressTaskCache;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.data.redis.core.RedisHash;

import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.concurrent.TimeUnit;

public class ProgressTaskCacheDaoImpl {

    /**

     * 默认分类.

     */

    private static final String KEY_SPACE = ProgressTaskCache.class.getAnnotation(RedisHash.class).value();

    /**

     * redis 操作.

     */

    @Autowired

    private StringRedisTemplate redisTemplate;

    /**

     * 是否过期.

     *

     * @param id    主键.

     * @param delay 延迟.

     * @return 布尔值.

     */

    public boolean expireAt(String id, long delay) {

        String key = KEY_SPACE + ":" + id;

        // 临时保存 delay 毫秒

        return redisTemplate.expire(key, delay, TimeUnit.MILLISECONDS);

    }

}

package com.example.demo.service;

import com.example.demo.dto.Type;

import com.example.demo.entity.ProgressTaskCache;

public interface ProgressTaskCacheService {

    Type PROPGRESS_STATUS_WAIT = new Type("WAIT", "等待");

    Type PROPGRESS_STATUS_DOING = new Type("DOING", "执行中");

    Type PROPGRESS_STATUS_COMPLETE = new Type("COMPLETE", "完成");

    Type PROPGRESS_STATUS_ERROR = new Type("ERROR", "失败");

    Type PROPGRESS_STATUS_CANCEL = new Type("CANCEL", "取消");

    /**

     * 开始异步任务

     *

     * @return

     */

    ProgressTaskCache startForSimple();

    /**

     * 获取消息任务信息。

     *

     * @param id

     * @return

     */

    ProgressTaskCache findById_(String id);

    /**

     * 完成任务。

     *

     * @param id

     * @return

     */

    void complete(String id);

    /**

     * 完成任务，填写消息。

     *

     * @param id

     * @param message

     */

    void complete(String id, String message);

    ProgressTaskCache save_(ProgressTaskCache tokenCache);

    /**

     * 获取进度任务情况，并清除完成状态的缓存数据

     *

     * @param id

     * @return

     */

    ProgressTaskCache getAndClearCompleteCache(String id);

}

package com.example.demo.service;

import com.example.demo.dao.ProgressTaskCacheDao;

import com.example.demo.entity.ProgressTaskCache;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

import java.util.UUID;

@Service

public class ProgressTaskCacheServiceImpl implements ProgressTaskCacheService {

    @Autowired

    private ProgressTaskCacheDao progressTaskCacheDao;

    @Override

    public ProgressTaskCache startForSimple() {

        return progressTaskCacheDao.save(create(null));

    }

    private ProgressTaskCache create(String id) {

        ProgressTaskCache progressTaskCache = new ProgressTaskCache();

        if (id == null) id = UUID.randomUUID().toString();

        progressTaskCache.setId(id);

        progressTaskCache.setStatus(PROPGRESS_STATUS_WAIT);

        return progressTaskCache;

    }

    @Override

    public void complete(String id) {

        complete(id,null);

    }

    @Override

    public ProgressTaskCache save_(ProgressTaskCache progressTaskCache) {

        return progressTaskCacheDao.save(progressTaskCache);

    }

    @Override

    public ProgressTaskCache findById_(String id) {

        return progressTaskCacheDao.findById(id).orElse(null);

    }

    @Override

    public void complete(String id, String message) {

        ProgressTaskCache progressTaskCache = progressTaskCacheDao.findById(id).orElse(null);

        if (progressTaskCache == null) return;

        progressTaskCache.setStatus(PROPGRESS_STATUS_COMPLETE);

        progressTaskCache.setMessage(message);

        progressTaskCacheDao.save(progressTaskCache);

        progressTaskCacheDao.expireAt(id, progressTaskCache.getTimeout());

    }

    /**

     * 获取进度任务情况，并清除完成状态的缓存数据

     *

     * @param id

     * @return

     */

    @Override

    public ProgressTaskCache getAndClearCompleteCache(String id) {

        ProgressTaskCache progressTaskCache = progressTaskCacheDao.findById(id).orElse(null);

        // 临时改为延长时间，后续有再调整.

        if (isComplete(progressTaskCache)) {

            // 已完成的缩短有效时间为 1 分钟。

            progressTaskCacheDao.expireAt(id, 60000);

        }

        return progressTaskCache;

    }

    // 是否完成状态

    private boolean isComplete(ProgressTaskCache progressTaskCache) {

        // 临时改为延长时间，后续有再调整.

        if (PROPGRESS_STATUS_COMPLETE.equals(progressTaskCache.getStatus())

                || PROPGRESS_STATUS_ERROR.equals(progressTaskCache.getStatus())

                || PROPGRESS_STATUS_CANCEL.equals(progressTaskCache.getStatus())) {

            return true;

        }

        return false;

    }

}

package com.example.demo.service;

import com.example.demo.dto.ResponseResult;

public interface DemoService {

    ResponseResult start_1();

}

package com.example.demo.service;

import com.example.demo.dao.ProgressTaskCacheDao;

import com.example.demo.dto.ResponseResult;

import com.example.demo.entity.ProgressTaskCache;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service

public class DemoServiceImpl implements DemoService {

    private static final Logger LOG = LoggerFactory.getLogger(DemoServiceImpl.class);

    @Autowired

    private ProgressTaskCacheService progressTaskCacheFacade;

    @Autowired

    private ProgressTaskCacheDao progressTaskCacheDao;

    @Override

    public ResponseResult start_1(){

        // 开始异步任务

        ProgressTaskCache progressTaskCache = progressTaskCacheFacade.startForSimple();

        ResponseResult responseCode = new ResponseResult();

        // 将异步任务信息返回

        responseCode.setProgressTask(progressTaskCache);

        LOG.info("异步任务开始...");

        CompletableFuture<ResponseResult> f = CompletableFuture.supplyAsync(this::demo).

                whenComplete((res,err) -> {

                    if (err != null) {

                        err.printStackTrace();

                    }

                    try {

                        if (res != null && res.getMsg() != null) {

                            progressTaskCache.setMessage(res.getMsg());

                        }

                    } catch (Exception e) {

                        e.printStackTrace();

                    }

                    progressTaskCacheFacade.complete(progressTaskCache.getId(), progressTaskCache.getMessage());

                    LOG.info("异步任务结束...");

                });

        // 返回

        return responseCode;

    }

    /**

     * 模拟某段逻辑长时间运行

     *

     * @return

     */

    private ResponseResult demo() {

        for (int i = 0; i < 30; i++) {

            LOG.info("处理业务{}", i + 1);

            try {

                Thread.sleep(1000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

        }

        ResponseResult responseResult = new ResponseResult();

        responseResult.setMsg("测试成功");

        responseResult.setCode(200);

        return responseResult;

    }

}

package com.example.demo.controller;

import com.example.demo.dto.ResponseResult;

import com.example.demo.entity.ProgressTaskCache;

import com.example.demo.service.DemoService;

import com.example.demo.service.ProgressTaskCacheService;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RequestParam;

import org.springframework.web.bind.annotation.RestController;

@RestController

public class DemoController {

    @Autowired

    private DemoService demoService;

    @Autowired

    private ProgressTaskCacheService progressTaskCacheService;

    @GetMapping("/start_1")

    private ResponseResult start_1(){

        return demoService.start_1();

    }

    @GetMapping("/getAndClearCompleteCache")

    private ProgressTaskCache getAndClearCompleteCache(@RequestParam("id") String id){

        return progressTaskCacheService.getAndClearCompleteCache(id);

    }

}

package com.example.demo.ansyc;

import com.example.demo.dto.ResponseResult;

import com.example.demo.entity.ProgressTaskCache;

import com.example.demo.service.ProgressTaskCacheService;

import com.example.demo.util.ConvertUtil;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.scheduling.annotation.Async;

import org.springframework.stereotype.Component;

import javax.validation.ValidationException;

@Component

@Async

public class DemoAsyncTask {

    private static final Logger LOG = LoggerFactory.getLogger(DemoAsyncTask.class);

    @Autowired

    private ProgressTaskCacheService progressTaskCacheService;

    public void importByExcel(ProgressTaskCache progressTaskCache){

        try {

            // 执行数据导入

            importByExcel_(progressTaskCache);

        } catch (Exception e) {

            // 导入失败

            progressTaskCache.setStatus(ProgressTaskCacheService.PROPGRESS_STATUS_ERROR);

            // 获取失败信息

            String message = getResolvedMessage(e);

            // 返回失败信息

            writeFromDataToProgressTaskCache(progressTaskCache, message);

        } finally {

            closeProgressTaskCache(progressTaskCache);

        }

    }

    /**

     * 获取异常信息

     *

     * @param e

     * @return

     */

    private String getResolvedMessage(Exception e) {

        if (e instanceof ValidationException) {

            return e.getMessage();

        } else {

            return e.getMessage();

        }

    }

    /**

     * 关闭异步任务

     *

     * @param mProgressTaskCache

     */

    private void closeProgressTaskCache(ProgressTaskCache mProgressTaskCache) {

        progressTaskCacheService.save_(mProgressTaskCache);

    }

    private void writeFromDataToProgressTaskCache(ProgressTaskCache mProgressTaskCache, String message) {

        ResponseResult responseResult = new ResponseResult();

        responseResult.setCode(400);

        responseResult.setMsg(message);

        mProgressTaskCache.setData(ConvertUtil.toJson(responseResult));

        // 无论如何需要更新结果

        mProgressTaskCache.setStatus(ProgressTaskCacheService.PROPGRESS_STATUS_ERROR);

    }

    private void importByExcel_(ProgressTaskCache progressTaskCache) {

        for (int i = 0; i < 15; i++) {

            LOG.info("开始导入第" + (i + 1) + "条数据");

            try {

                Thread.sleep(1000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

        }

        ResponseResult responseCode = new ResponseResult();

        responseCode.setMsg("导入成功");

        responseCode.setCode(200);

        progressTaskCache.setData(ConvertUtil.toJson(responseCode));

        // 无论如何需要更新结果

        progressTaskCache.setStatus(ProgressTaskCacheService.PROPGRESS_STATUS_COMPLETE);

    }

}

package com.example.demo.service;

import com.example.demo.dto.ResponseResult;

public interface DemoService {

    // ...

    ResponseResult start_2();

}

package com.example.demo.service;

import com.example.demo.ansyc.DemoAsyncTask;

import com.example.demo.dao.ProgressTaskCacheDao;

import com.example.demo.dto.ResponseResult;

import com.example.demo.entity.ProgressTaskCache;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service

public class DemoServiceImpl implements DemoService {

    private static final Logger LOG = LoggerFactory.getLogger(DemoServiceImpl.class);

    @Autowired

    private ProgressTaskCacheService progressTaskCacheService;

    @Autowired

    private ProgressTaskCacheDao progressTaskCacheDao;

    @Autowired

    private DemoAsyncTask demoAsyncTask;

    // ...

    @Override

    public ResponseResult start_2() {

        ProgressTaskCache mProgressTaskCache = progressTaskCacheService.startForSimple();

        // 委派给异步组件进行处理

        demoAsyncTask.importByExcel(mProgressTaskCache);

        ResponseResult result = new ResponseResult();

        result.setProgressTask(mProgressTaskCache);

        return result;

    }

}

package com.example.demo.controller;

import com.example.demo.dto.ResponseResult;

import com.example.demo.entity.ProgressTaskCache;

import com.example.demo.service.DemoService;

import com.example.demo.service.ProgressTaskCacheService;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RequestParam;

import org.springframework.web.bind.annotation.RestController;

@RestController

public class DemoController {

    @Autowired

    private DemoService demoService;

    @Autowired

    private ProgressTaskCacheService progressTaskCacheService;

	// ...

    @GetMapping("/start_2")

    private ResponseResult start_2(){

        return demoService.start_2();

    }

    @GetMapping("/getAndClearCompleteCache")

    private ProgressTaskCache getAndClearCompleteCache(@RequestParam("id") String id){

        return progressTaskCacheService.getAndClearCompleteCache(id);

    }

}