如何在Java Microprofile中正确实现异步POST/GET请求处理

核心问题解答 #

POST接口需要返回查询URI。根据HTTP 202 Accepted状态码的规范约定，服务端需要在响应头的Location字段返回异步任务的状态查询URI，客户端直接使用该地址轮询即可，无需自行拼接参数。

你之前尝试的AsyncResponse不适用当前场景：该组件用于异步处理完成后再给客户端返回响应的场景，和你要求POST立即返回202、业务逻辑后台异步执行的需求不匹配。

实现步骤 #

1. 前置准备：线程安全的任务状态存储 #

首先需要全局线程安全的容器存储任务状态、进度、结果，避免多线程读写冲突。Microprofile支持CDI依赖注入，直接定义应用级别的存储组件即可：

import jakarta.enterprise.context.ApplicationScoped;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

// 任务状态枚举
enum TaskStatus {
    RUNNING, SUCCESS, FAILED
}

// 任务信息实体，Java 16+可用record，低版本改用普通类
record TaskInfo(
    String taskId,
    TaskStatus status,
    int progress, // 0-100 百分比
    Object result // 处理完成后的返回结果
) {}

@ApplicationScoped
public class TaskStore {
    private final ConcurrentMap<String, TaskInfo> taskMap = new ConcurrentHashMap<>();

    // 新增任务
    public void addTask(TaskInfo task) {
        taskMap.put(task.taskId(), task);
    }

    // 更新任务进度/状态
    public void updateTask(String taskId, TaskStatus status, int progress, Object result) {
        taskMap.put(taskId, new TaskInfo(taskId, status, progress, result));
    }

    // 查询任务
    public TaskInfo getTask(String taskId) {
        return taskMap.get(taskId);
    }
}

2. POST接口实现 #

Microprofile规范提供容器托管的ManagedExecutorService，直接注入即可使用，不需要自己创建线程池，避免资源泄漏：

import jakarta.enterprise.context.RequestScoped;
import jakarta.inject.Inject;
import jakarta.ws.rs.*;
import jakarta.ws.rs.core.*;
import org.eclipse.microprofile.context.ManagedExecutorService;
import java.util.UUID;

@Path("/logic")
@RequestScoped
public class LogicResource {
    @Inject
    ManagedExecutorService executor;

    @Inject
    TaskStore taskStore;

    @Context
    UriInfo uriInfo;

    @POST
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    public Response logic(JsonObject jsonObject) {
        // 生成唯一任务ID
        String taskId = UUID.randomUUID().toString();
        // 初始化任务状态
        taskStore.addTask(new TaskInfo(taskId, TaskStatus.RUNNING, 0, null));

        // 提交业务逻辑到托管线程池异步执行
        executor.submit(() -> {
            BusinessLogic bl = new BusinessLogic();
            // 业务逻辑内部调用taskStore.updateTask更新进度
            bl.doSomethingWithInput(jsonObject, taskId, taskStore);
            // 执行完成后更新状态为SUCCESS，写入结果
            // 如果执行异常更新为FAILED
        });

        // 构建查询URI，写入Location响应头
        URI queryUri = uriInfo.getAbsolutePathBuilder()
                .path("../logicstatus/" + taskId)
                .build();
        return Response.status(Response.Status.ACCEPTED)
                .header(HttpHeaders.LOCATION, queryUri.toString())
                .build();
    }
}

业务逻辑类中进度更新示例：

public class BusinessLogic {
    public void doSomethingWithInput(JsonObject input, String taskId, TaskStore taskStore) {
        // 步骤1处理，进度更新为20%
        taskStore.updateTask(taskId, TaskStatus.RUNNING, 20, null);
        // ... 步骤1逻辑

        // 步骤2处理，进度更新为60%
        taskStore.updateTask(taskId, TaskStatus.RUNNING, 60, null);
        // ... 步骤2逻辑

        // 处理完成，写入结果
        Object result = "你的处理结果";
        taskStore.updateTask(taskId, TaskStatus.SUCCESS, 100, result);
    }
}

3. GET接口实现 #

直接使用路径参数接收任务ID即可，无需把整个URI作为查询参数：

import jakarta.inject.Inject;
import jakarta.ws.rs.*;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;

@Path("/logicstatus")
public class LogicStatusResource {
    @Inject
    TaskStore taskStore;

    @GET
    @Path("/{taskId}")
    @Produces(MediaType.APPLICATION_JSON)
    public Response logicStatus(@PathParam("taskId") String taskId) {
        TaskInfo task = taskStore.getTask(taskId);
        if (task == null) {
            return Response.status(Response.Status.NOT_FOUND).build();
        }
        // 自定义返回结构，也可以直接返回TaskInfo对象
        JsonObject result = Json.createObjectBuilder()
                .add("status", task.status().name())
                .add("progress", task.progress())
                .add("result", task.result() == null ? "" : task.result().toString())
                .build();
        return Response.ok(result).build();
    }
}

注意事项 #

• 如果服务端是多实例部署，需要把TaskStore的存储实现换成Redis、数据库等共享存储，单实例部署用ConcurrentHashMap即可
• 可根据业务需求添加任务过期清理逻辑，避免存储溢出
• 线程池参数可以通过Microprofile Config配置，无需修改代码

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