servlet - 管理同时的服务器推送通知

Question

我正在设计一个非常简单的聊天应用程序，它使用长轮询在用户之间进行通信。我正在使用 servlet 将 AsyncContexts 存储在 Hashmap 中。当 userA 向 userB 发送消息时，servlet 从 HashMap 中获取相应的 AsyncContext 并将数据推送到客户端。

我使用的代码如下：

protected void doGet(HttpServletRequest request,
            HttpServletResponse response) throws ServletException, IOException {

        response.setContentType("application/json");
        if ("getAll".equals(request.getParameter("op"))) {
            String friend = request.getParameter("for");
            String jsonMessages = getChatMessagesForUser((String)request.getSession()
                    .getAttribute("id"),friend);
            PrintWriter writer = response.getWriter();
            writer.println(jsonMessages);
            writer.flush();
            return;
        }

        else if ("getFrnd".equals(request.getParameter("op"))) {
            String jsonFrndList = getFriends((String)request.getSession()
                    .getAttribute("id"));
            PrintWriter writer = response.getWriter();
            writer.println(jsonFrndList);
            writer.flush();
            return;
        }

        final AsyncContext asyncContext = request.startAsync(request, response);
        asyncContext.setTimeout(30 * 60 * 1000);
        asyncContext.addListener(new AsyncListener() {

            @Override
            public void onTimeout(AsyncEvent event) throws IOException {
                AsyncContext ctx = event.getAsyncContext();
                PrintWriter writer = ctx.getResponse().getWriter();
                ChatMessage directive = new ChatMessage();
                directive.setType('D');
                directive.setMessage("Keep-Alive");
                writer.println(gson.toJson(directive));
                writer.flush();
                contexts.remove(ctx);
                ctx.complete();
            }

            @Override
            public void onStartAsync(AsyncEvent event) throws IOException {
                // TODO Auto-generated method stub

            }

            @Override
            public void onError(AsyncEvent event) throws IOException {
                AsyncContext ctx = event.getAsyncContext();
                PrintWriter writer = ctx.getResponse().getWriter();
                ChatMessage directive = new ChatMessage();
                directive.setType('D');
                directive.setMessage("Keep-Alive");
                writer.println(gson.toJson(directive));
                writer.flush();
                contexts.remove(ctx);
                ctx.complete();

            }

            @Override
            public void onComplete(AsyncEvent event) throws IOException {
                // TODO Auto-generated method stub

            }
        });
        String id = (String) request.getSession().getAttribute("id");

        if(!contexts.containsKey(id)) {
            contexts.put(id, asyncContext);
        }
    }
protected void doPost(HttpServletRequest request,
            HttpServletResponse response) throws ServletException, IOException {

        response.setContentType("application/json");
        String receiver = request.getParameter("name");
        String sender = (String) request.getSession().getAttribute("id");
        String message = request.getParameter("message");

        System.out.println(receiver + " " + message);

        ChatMessage chatMessage = new ChatMessage();

        chatMessage.setType('M');
        chatMessage.setReceiver(receiver);
        chatMessage.setSender(sender);
        chatMessage.setMessage(message);
        chatMessage.setImage(false);

        if(contexts.containsKey(receiver)) {

            AsyncContext recieverContext = (AsyncContext) contexts.remove(receiver);
            System.out.println("Connections Open:" + contexts.size());
            PrintWriter writer = recieverContext.getResponse().getWriter();
            writer.println(gson.toJson(chatMessage));
            System.out.println("Message transmitted:" + 
            gson.toJson(chatMessage));
            writer.flush();
            recieverContext.complete();
        }

        try {
            System.out.println("Trying to add to database");
            ChatDAO.addToChat(chatMessage);
        } catch (SQLException e) {
            System.err.println(e);
        }

    }

让我解释一下发生了什么-：

• 用户 A 向 servlet 发送初始 GET 请求。
• servlet 启动 AsyncContext 并将其添加到 HashMap。
• 用户 B 做同样的事情。
• 用户 A 通过发布到 servlet(doPost) 向用户 B 发送消息。
• servlet 从 hashmap 中获取用户 B 的 AsyncContext。推送来自用户 A 的消息并对其调用 complete()。
• 用户 B 通过执行另一个 GET (doGet) 立即重新连接到 servlet。

这个明显的问题是当多人尝试与一个人聊天时。 例如-：如果用户 B 和用户 C 尝试与用户 A 聊天，则会出现竞争条件。

可能会发生他们同时向用户 A 发布消息，这取决于首先处理谁的消息，与用户 A 的连接将被关闭。 因此，如果聊天消息在收件人重新连接之前到达，则该消息将被丢弃。

我需要的是，如果 servlet 无法为特定用户找到异步上下文，那么它将等待一段时间，然后在等待之后如果仍然找不到它，它将丢弃该消息。

我可以写类似 -:

synchronized(contexts) {
wait(100);
}

Answer 1

最好将消息的存储与异步上下文分开，这样即使没有接收者的上下文也可以存储消息。下次收件人发送 get 请求时，它会发现一条消息正在等待并立即返回。

如果您担心过时的未传递消息，那么您可以使用一个线程定期检查超过一定期限的消息并将它们返回给发件人。它也可能只是破坏旧的返回发件人消息