java - Jersey/JAX-RS 2 AsyncResponse - 如何跟踪当前的长轮询调用者

Question

我的目标是支持多个 Web 服务调用者的长轮询，并跟踪哪些调用者当前“停留”在长轮询上(即已连接)。我所说的“长轮询”是指调用者调用 Web 服务，而服务器(Web 服务)不会立即返回，而是让调用者等待一段预设的时间(在我的应用程序中为一个小时)，或者更快地返回如果服务器有消息要发送给调用者(在这种情况下，服务器通过调用 asyncResponse.resume("MESSAGE") 返回消息)。

我将把这个问题分成两个问题。

第一个问题:这是“停放”长轮询调用者的合理方式吗？

@GET
@Produces(MediaType.TEXT_PLAIN)
@ManagedAsync
@Path("/poll/{id}")
public Response poller(@Suspended final AsyncResponse asyncResponse, @PathParam("id") String callerId) {

    // add this asyncResponse to a HashMap that is persisted across web service calls by Jersey.
    // other application components that may have a message to send to a caller will look up the
    // caller by callerId in this HashMap and call resume() on its asyncResponse.
    callerIdAsyncResponseHashMap.put(callerId, asyncResponse);

    asyncResponse.setTimeout(3600, TimeUnit.SECONDS);
    asyncResponse.setTimeoutHandler(new TimeoutHandler() {
        @Override
        public void handleTimeout(AsyncResponse asyncResponse) {
            asyncResponse.resume(Response.ok("TIMEOUT").build());
        }
    });
    return Response.ok("COMPLETE").build();
}

这很好用。我只是不确定它是否遵循最佳实践。方法末尾有“return Response...”行似乎很奇怪。该行在调用者首次连接时执行，但是据我了解，“完整”结果实际上从未返回给调用者。当服务器需要通知调用者某个事件时，调用者要么获得“TIMEOUT”响应，要么获得服务器通过 asyncResponse.resume() 发送的其他响应消息。

第二个问题:我当前的挑战是在 HashMap 中准确反射(reflect)当前轮询调用者的数量。当调用者停止轮询时，我需要从 HashMap 中删除其条目。调用者可以出于以下三个原因离开:1) 3600 秒过去，因此超时，2) 另一个应用程序组件在 HashMap 中查找调用者并调用 asyncResponse.resume("MESSAGE")，以及 3) HTTP 连接已关闭由于某种原因损坏，例如有人关闭了运行客户端应用程序的计算机。

因此，JAX-RS 有两个回调，我可以注册以接收连接结束的通知:CompletionCallback(用于我上面的结束轮询原因 #1 和 #2)和 ConnectionCallback(用于我上面的结束轮询原因 #3) .

我可以将这些添加到我的网络服务方法中，如下所示:

@GET
@Produces(MediaType.TEXT_PLAIN)
@ManagedAsync
@Path("/poll/{id}")
public Response poller(@Suspended final AsyncResponse asyncResponse, @PathParam("id") String callerId) {

    asyncResponse.register(new CompletionCallback() {
        @Override
        public void onComplete(Throwable throwable) {
            //?
        }
    });

    asyncResponse.register(new ConnectionCallback() {
        @Override
        public void onDisconnect(AsyncResponse disconnected) {
            //?
        }
    });

    // add this asyncResponse to a HashMap that is persisted across web service calls by Jersey.
    // other application components that may have a message to send to a caller will look up the
    // caller by callerId in this HashMap and call resume() on its asyncResponse.
    callerIdAsyncResponseHashMap.put(callerId, asyncResponse);

    asyncResponse.setTimeout(3600, TimeUnit.SECONDS);
    asyncResponse.setTimeoutHandler(new TimeoutHandler() {
        @Override
        public void handleTimeout(AsyncResponse asyncResponse) {
            asyncResponse.resume(Response.ok("TIMEOUT").build());
        }
    });
    return Response.ok("COMPLETE").build();
}

正如我所说，挑战是使用这两个回调从 HashMap 中删除不再轮询的调用者。 ConnectionCallback 实际上是两者中更容易的一个。由于它接收 asyncResponse 实例作为参数，我可以使用它从 HashMap 中删除相应的条目，如下所示:

asyncResponse.register(new ConnectionCallback() {
    @Override
    public void onDisconnect(AsyncResponse disconnected) {
        Iterator<Map.Entry<String, AsyncResponse>> iterator = callerIdAsyncResponseHashMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, AsyncResponse> entry = iterator.next();
            if (entry.getValue().equals(disconnected)) {
                iterator.remove();
                break;
            }
        }
    }
});

但是，对于 CompletionCallback，由于在触发回调时 asyncResponse 已经完成或取消，因此没有传入 asyncResponse 参数。因此，似乎唯一的解决方案是运行 HashMap 条目检查完成/取消的并删除它们，如下所示。 (请注意，我不需要知道调用者离开是因为调用了resume() 还是因为超时，因此我不查看“可抛出”参数)。

asyncResponse.register(new CompletionCallback() {
    @Override
    public void onComplete(Throwable throwable) {
        Iterator<Map.Entry<String, AsyncResponse>> iterator = callerIdAsyncResponseHashMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, AsyncResponse> entry = iterator.next();
            if (entry.getValue().isDone() || entry.getValue().isCancelled()) {
                iterator.remove();
            }
        }
    }
});

如有任何反馈，我们将不胜感激。这种做法看起来合理吗？有更好或更多种 Jersey/JAX-RS 方法吗？

Answer 1

您的 poller() 方法不需要返回 Response 即可参与异步处理。它可以返回 void。但是，如果您在轮询器中执行任何复杂的操作，则应考虑将整个方法包装在 try/catch block 中，该 block 将恢复 AsyncResponse 对象，但会出现异常，以确保任何 RuntimeException 或其他未经检查的 Throwable 不会丢失。将这些异常记录在此处的 catch block 中似乎也是一个好主意。

我目前正在研究如何可靠地捕获客户端取消的异步请求的问题，并阅读了一个问题，该问题表明该机制不适用于提问者[1]。我暂时将这些信息留给其他人来填写。

[1] AsyncResponse ConnectionCallback does not fire in Jersey