Java异步任务完成/进度监视器与用户会话关联实现方案咨询

你遇到的核心问题是AsyncResult不可序列化没法存会话，那我们可以绕开它，自定义可序列化的状态追踪对象，再结合全局存储或会话关联来实现任务进度/结果的追踪。下面是具体的实现步骤：

1. 自定义可序列化的任务状态对象 #

首先创建一个实现Serializable的类，用来存储任务的核心状态信息，这样就能安全地存入HttpSession或者全局存储中：

import java.io.Serializable;
import java.util.concurrent.atomic.AtomicInteger;

public class TaskStatus implements Serializable {
    private static final long serialVersionUID = 1L;
    
    private String taskId;
    private String status; // 可选值：PENDING, RUNNING, COMPLETED, FAILED
    private AtomicInteger progress; // 用Atomic类保证线程安全
    private Object result;
    private String errorMessage;

    // 构造器、getter、setter方法
    public TaskStatus(String taskId) {
        this.taskId = taskId;
        this.status = "PENDING";
        this.progress = new AtomicInteger(0);
    }

    // 省略getter和setter
}

2. 全局任务状态存储（替代直接存会话） #

因为无状态EJB不能直接访问HttpSession，我们可以用一个线程安全的全局存储来管理所有任务的状态，比如ConcurrentHashMap，这样EJB和JAX-RS端点都能通过任务ID来访问状态：

import javax.ejb.Singleton;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

@Singleton
public class TaskStatusStore {
    private final Map<String, TaskStatus> taskMap = new ConcurrentHashMap<>();

    public String createNewTask() {
        String taskId = UUID.randomUUID().toString();
        taskMap.put(taskId, new TaskStatus(taskId));
        return taskId;
    }

    public TaskStatus getTaskStatus(String taskId) {
        return taskMap.get(taskId);
    }

    public void removeTask(String taskId) {
        taskMap.remove(taskId);
    }
}

3. 修改无状态EJB为异步方法，更新任务状态 #

改造原来的同步EJB方法为异步，注入TaskStatusStore，在任务执行过程中更新进度和状态：

import javax.ejb.Asynchronous;
import javax.ejb.Stateless;
import javax.inject.Inject;

@Stateless
public class AsyncTaskEJB {

    @Inject
    private TaskStatusStore taskStatusStore;

    @Asynchronous
    public void executeLongRunningTask(String taskId) {
        TaskStatus status = taskStatusStore.getTaskStatus(taskId);
        status.setStatus("RUNNING");

        try {
            // 模拟长时间任务，分步骤执行并更新进度
            for (int i = 0; i <= 100; i += 10) {
                Thread.sleep(500); // 模拟任务耗时
                status.getProgress().set(i);
            }

            status.setStatus("COMPLETED");
            status.setResult("任务执行完成！");
        } catch (InterruptedException e) {
            status.setStatus("FAILED");
            status.setErrorMessage("任务被中断：" + e.getMessage());
            Thread.currentThread().interrupt();
        }
    }
}

4. JAX-RS端点处理会话关联与任务交互 #

JAX-RS端点负责发起任务、关联用户会话，并提供查询进度/结果的接口：

import javax.inject.Inject;
import javax.servlet.http.HttpSession;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.core.Response;

@Path("/tasks")
public class TaskResource {

    @Inject
    private AsyncTaskEJB asyncTaskEJB;

    @Inject
    private TaskStatusStore taskStatusStore;

    @POST
    @Path("/start")
    public Response startTask(HttpSession session) {
        // 创建新任务并获取ID
        String taskId = taskStatusStore.createNewTask();
        // 将任务ID存入用户会话，关联当前用户
        session.setAttribute("currentTaskId", taskId);
        // 触发异步任务执行
        asyncTaskEJB.executeLongRunningTask(taskId);
        return Response.ok(taskId).build();
    }

    @GET
    @Path("/status")
    public Response getTaskStatus(HttpSession session) {
        String taskId = (String) session.getAttribute("currentTaskId");
        if (taskId == null) {
            return Response.status(Response.Status.BAD_REQUEST).entity("无正在执行的任务").build();
        }

        TaskStatus status = taskStatusStore.getTaskStatus(taskId);
        if (status == null) {
            return Response.status(Response.Status.NOT_FOUND).entity("任务不存在").build();
        }

        // 返回任务状态，这里可以转成JSON
        return Response.ok(status).build();
    }

    @POST
    @Path("/cleanup")
    public Response cleanupTask(HttpSession session) {
        String taskId = (String) session.getAttribute("currentTaskId");
        if (taskId != null) {
            taskStatusStore.removeTask(taskId);
            session.removeAttribute("currentTaskId");
        }
        return Response.ok("任务已清理").build();
    }
}

5. 进阶优化：用WebSocket主动推送进度 #

如果觉得轮询不够高效，可以用WebSocket替代，当任务进度更新时主动推送给前端，这样用户体验更好：

• 建立WebSocket连接时，将用户会话与WebSocket会话关联
• EJB更新任务状态时，通过WebSocket会话主动推送进度信息

注意事项 #

• 线程安全：任务状态对象的字段要保证线程安全，比如用AtomicInteger存进度，或者对字段更新加锁
• 内存清理：任务完成后要及时从全局存储和会话中移除状态对象，避免内存泄漏
• 异常处理：要捕获任务执行中的异常，更新状态为失败并记录错误信息
• 会话超时：如果用户会话超时，未完成的任务可以选择继续执行或终止，根据业务需求处理

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