Dropwizard 中的 JAX-RS：处理异步调用并立即响应

Question

我有一个资源Class，有一个@ManagedAsync 方法类，看起来像这样：

@Path("my-resource")
public class MyResource extends BaseResource{

    private DatumDAO datumDAO;

    public MyResource(DatumDAO datumDAO){
        this.datumDAO = datumDAO;
    }

    public void cleanDatum(Datum datum){
       //time taking operations
    }

    @GET
    @ManagedAsync
    @Path("/cleanup/{from}/{till}/")
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    @UnitOfWork
    public void cleanupDirtyData(@Suspended final AsyncResponse asyncResponse, @PathParam("from") DateTimeParam from,
            @PathParam("till") DateTimeParam till) throws IOException{

        logger.debug("will try to cleanup dirty data in range: " + from + " " +  till);
        List<Datum> data = datumDAO.getALlDirtyDatumInRange(from.get().toDate(), till.get().toDate());
        Map<Long,String> cleanupMap = new HashMap<Long,String>();
        for(Datum datum: data){
            cleanDatum(datum);
            cleanupMap.put(datum.getId(), "cleaned");
        }
        // this response need to be sent [can be ignored]       
        asyncResponse.resume(Response.status(HttpStatus.OK_200).entity(cleanupMap).build());

    }

}

由于对cleanupDirtyData 的调用可能需要一段时间，我不希望客户端完全等待它，我知道执行工作已卸载到不同的工作线程。

我想要实现的是立即响应客户端并继续异步执行函数cleanupDirtyData。

于是尝试了以下方法：

强制超时，并过早地响应客户端，但这似乎不是理想的方式，它会停止执行。

看起来像这样：

@Path("my-resource")
public class MyResource extends BaseResource{

    private DatumDAO datumDAO;

    public MyResource(DatumDAO datumDAO){
        this.datumDAO = datumDAO;
    }

    public void cleanDatum(Datum datum){
       //time taking operations
    }

    @GET
    @ManagedAsync
    @Path("/cleanup/{from}/{till}/")
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    @UnitOfWork
    public void cleanupDirtyData(@Suspended final AsyncResponse asyncResponse, @PathParam("from") DateTimeParam from,
            @PathParam("till") DateTimeParam till) throws IOException{

        // Register handler and set timeout
        asyncResponse.setTimeoutHandler(new TimeoutHandler() {
            public void handleTimeout(AsyncResponse ar) {
                asyncResponse.resume(Response.status(SERVICE_UNAVAILABLE).entity(
                    "Operation timed out -- please try again").build());                    
                }
        });
        ar.setTimeout(15, TimeUnit.SECONDS);      

        logger.debug("will try to cleanup dirty data in range: " + from + " " +  till);
        List<Datum> data = datumDAO.getALlDirtyDatumInRange(from.get().toDate(), till.get().toDate());
        Map<Long,String> cleanupMap = new HashMap<Long,String>();
        for(Datum datum: data){
            cleanDatum(datum);
            cleanupMap.put(datum.getId(), "cleaned");
        }
       // this response need to be sent [can be ignored]              
        asyncResponse.resume(Response.status(HttpStatus.OK_200).entity(cleanupMap).build());

    }

}

Answer 1

JAX-RS 异步服务器 API 是关于 容器 将如何管理请求的全部内容。但它仍会保留请求，不会影响客户端体验。

引用关于 Asynchronous Server API 的 Jersey 文档：

注意使用服务器端异步处理模型会 不会改善客户端感知的请求处理时间。它 然而，将增加服务器的吞吐量，通过释放 初始请求处理线程返回到 I/O 容器，而 请求可能仍在队列中等待处理或 处理可能仍在另一个专用线程上运行。这 释放的 I/O 容器线程可用于接受和处理新的 传入请求连接。

如果您想立即给客户回复，您可能正在寻找类似的东西：

@Singleton
@Path("expensive-task")
public class ExpensiveTaskResource {

    private ExecutorService executor;

    private Future<String> futureResult;

    @PostConstruct
    public void onCreate() {
        this.executor = Executors.newSingleThreadExecutor();
    }

    @POST
    public Response startTask() {
        futureResult = executor.submit(new ExpensiveTask());
        return Response.status(Status.ACCEPTED).build();
    }

    @GET
    public Response getResult() throws ExecutionException, InterruptedException {
        if (futureResult != null && futureResult.isDone()) {
            return Response.status(Status.OK).entity(futureResult.get()).build();
        } else {
            return Response.status(Status.FORBIDDEN).entity("Try later").build();
        }
    }

    @PreDestroy
    public void onDestroy() {
        this.executor.shutdownNow();
    }
}

public class ExpensiveTask implements Callable<String> {

    @Override
    public String call() throws Exception {

        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return "Task completed";
    }
}

在 servlet 容器中，您可以使用 ExecutorService 来运行昂贵的任务。在 Java EE 容器中，您应该考虑使用 ManagedExecutorService。