如何跟踪在多个服务器中运行的异步任务的进度状态答案

【问题标题】：How to track progress status of async tasks running in multiple servers如何跟踪在多个服务器中运行的异步任务的进度状态
【发布时间】：2019-09-12 13:07:43
【问题描述】：

我在 spring boot 中运行了多个异步任务。这些任务读取一个 excel 文件并将所有数据插入到数据库中。

任务在前端发出请求时启动。然后前端会周期性地不断请求任务的进度状态。

我需要跟踪每项任务的进度并知道它们何时完成。

这是接收任务请求并轮询其进度状态的控制器文件：

public class TaskController {

    @RequestMapping(method = RequestMethod.POST, value = "/uploadExcel")
    public ResponseEntity<?> uploadExcel(String excelFilePath) {
        String taskId = UUID.randomUUID().toString();
        taskAsyncService.AsyncManager(id, excelFilePath);

        HashMap<String, String> responseMap = new HashMap<>();
        responeMap.put("taskId",taskId);
        return new ResponseEntity<>(responseMap, HttpStatus.ACCEPTED);
    }

    // This will be polled to get progress of tasks being executed
    @RequestMapping(method = RequestMethod.GET, value = "/tasks/progress/{id}")
    public ResponseEntity<?> getTaskProgress(@PathVariable String taskId) {
        HashMap<String, String> map = new HashMap<>();

        if (taskAsyncService.containsTaskEntry(id) == null) {
            map.put("Error", "TaskId does not exist");
            return new ResponseEntity<>(map, HttpStatus.BAD_REQUEST);
        }

        boolean taskProgress = taskAsyncService.getTaskProgress(taskId);

        if (taskProgress) {
            map.put("message", "Task complete");
            taskAsyncService.removeTaskProgressEntry(taskId);
            return new ResponseEntity<>(map, HttpStatus.OK);
        }

        //Otherwise task is still running
        map.put("progressStatus", "Task running");
        return new ResponseEntity<>(map, HttpStatus.PARTIAL_CONTENT);

    }
}

这是执行异步任务的代码。

public class TaskAsyncService {
    private final AtomicReference<ConcurrentHashMap<String, Boolean>> isTaskCompleteMap = new AtomicReference<ConcurrentHashMap<String, Boolean>>();

    protected boolean containsTaskEntry(String taskId) {
        if (isTaskCompleteMap.get().get(taskId) != null) {
            return true;
        }
        return false;
    }

    protected boolean getTaskProgress(String taskId, String excelFilePath) {
        return isTaskCompleteMap.get().get(taskId);
    }

    protected void removeTaskProgressEntry(String taskId) {
        if (isTaskCompleteMap.get() != null) {
            isTaskCompleteMap.get().remove(taskId);
        }
    }

    @Async
    public CompletableFuture<?> AsyncManager(String taskId) {
        HashMap<String, String> map = new HashMap<>();

        //Add a new entry into isTaskCompleteMap
        isTaskCompleteMap.get().put(taskId, false);

        //Insert excel rows into database

        //Task completed set value to true
        isTaskCompleteMap.get().put(taskId, true);
        map.put("Success", "Task completed");

        return CompletableFuture.completedFuture(map);
    }
}

我正在使用带有负载均衡器的 AWS EC2。因此，有时一个轮询请求由新生成的服务器处理，该服务器不能访问 isTaskCompleteMap 并返回“TaskId 不存在”。

在这种情况下如何跟踪任务的状态？我知道我需要一个分布式数据结构，但不知道是什么类型以及如何实现它。

【问题讨论】：

标签： java spring-boot asynchronous amazon-ec2 distributed-computing

【解决方案1】：

您可以使用 Hazelcast 或类似的分布式解决方案（Redis 等）。

地图 - https://docs.hazelcast.org/docs/3.0/manual/html/ch02.html#Map

使用 hazelcast 而不是 CHM 中的分布式地图。
从此类地图获取应该返回任务，即使它们正在另一个 pod（服务器）上处理

【讨论】：