Tomcat - 服务器在 Tomcat8.5 上使用没有异步 servlet 的 HTTP2 发送事件

Question

我们在使用 AsyncServlets 时遇到了一些问题，因此我们研究了 SSE 和 WebSockets。

SSE 的所有 tomcat 示例如下所示:

@WebServlet("/TestServlet")
public class TestServlet extends HttpServlet {

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

        //content type must be set to text/event-stream
        response.setContentType("text/event-stream"); 
        //cache must be set to no-cache
        response.setHeader("Cache-Control", "no-cache");     
        //encoding is set to UTF-8
        response.setCharacterEncoding("UTF-8");

        PrintWriter writer = response.getWriter();

        for(int i=0; i<10; i++) {
            System.out.println(i);
            writer.write("data: "+ i +"\n\n");
            writer.flush();
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        writer.close(); 
    }
}

但是，我想长时间保持连接，希望不必让处理请求的线程保持事件状态。下面是从 PubSubHub 接收信息并希望将其推送到浏览器的示例。

@WebServlet("/TestServlet")
public class TestServlet extends HttpServlet {

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

        //content type must be set to text/event-stream
        response.setContentType("text/event-stream"); 
        //cache must be set to no-cache
        response.setHeader("Cache-Control", "no-cache");     
        //encoding is set to UTF-8
        response.setCharacterEncoding("UTF-8");

        final PrintWriter writer = response.getWriter();

        PublishSubsribeHub.getInstance().subscribe("TestData", new Callback(String msg){
           public void run(){
               writer.write("data:"+msg);
               writer.flush(); // this throws NullPointerException since undeline HttpServletResponse is closed
           });
}

如上面评论中所述，当数据准备就绪并通过 PublishSubsribeHub 接收时，连接已关闭。

“修复”此问题的方法是使用 response.startAsync() 保持连接打开。

但是，我们的问题恰恰在于这种机制“asyncContext”/“asyncServlet”。所以想知道(在我切换到 Tomcat 的 WebSocket 实现之前)是否有任何其他方法可以使用 Tomcat8.5 或 Tomcat9 支持异步推送消息，也许使用 HTTP/2 规范。

谢谢

Answer 1

如果在您的项目上下文中可行，请考虑使用 Jersey library为了启用 SSE functionality .

Jersey 是一种流行、易于使用且有据可查的解决方案。此外，该文档还包含有关如何使用这些库的清晰示例。您可以在下面找到这样的示例。

广播示例

import org.glassfish.jersey.media.sse.SseBroadcaster;

@Singleton
@Path("broadcast")
public static class BroadcasterResource {
    private SseBroadcaster broadcaster = new SseBroadcaster();

    @POST
    @Produces(MediaType.TEXT_PLAIN)
    @Consumes(MediaType.TEXT_PLAIN)
    public String broadcastMessage(String message) {
        OutboundEvent.Builder eventBuilder = new OutboundEvent.Builder();
        OutboundEvent event = eventBuilder.name("message")
            .mediaType(MediaType.TEXT_PLAIN_TYPE)
            .data(String.class, message)
            .build();

        broadcaster.broadcast(event);

        return "Message '" + message + "' has been broadcast.";
    }

    @GET
    @Produces(SseFeature.SERVER_SENT_EVENTS)
    public EventOutput listenToBroadcast() {
        final EventOutput eventOutput = new EventOutput();
        this.broadcaster.add(eventOutput);
        return eventOutput;
    }
}

这个具体的例子使用了一个广播:

• 允许客户端通过“listenToBroadcast”方法进行订阅。
• 使用“broadcastMessage”方法向每个订阅的客户端广播消息。

运行示例

为了运行示例，请按照以下步骤操作:

• 为 SSE 类创建一个新包
• 在此包中，根据上述Broadcaster 示例
创建一个类
• 添加另一个能够自动发现(和注册)存在于同一包中的每个类的类:

-

import javax.ws.rs.ApplicationPath;
import javax.ws.rs.core.Application;

@ApplicationPath("api")
public class ApiConfig extends Application { 

}

---

上述解决方案已使用 Tomcat 7、8、8.5 进行测试。使用的 Jersey 版本:2.x