使用RESTEasy结合Jetty实现Server-Sent Events(SSE)时线程池耗尽问题的解决方案咨询

EDIT:

• 替换了Thread.sleep()，改用线程池执行器实现
• 添加了servletHolder.setAsyncSupported(true)的调用

Question #

我们尝试用以下技术栈实现Server-Sent Events(SSE)：

• RESTEasy (6.2.11.Final)
• Jetty (12.0.15)

但遇到了一个头疼的问题：每个SSE订阅都会占用Jetty线程池中的一个线程，并且在整个订阅周期内一直持有不释放。

如果我们把RESTEasy换成Jersey，这个问题就完全不存在了，一切都正常工作。

有没有办法配置RESTEasy，避免它耗尽Jetty的线程池呢？

用RESTEasy实现SSE的代码 #

处理SSE订阅的资源类 #

这个资源类负责处理SSE订阅请求，每秒向订阅者推送一次事件：

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.Context;
import jakarta.ws.rs.sse.OutboundSseEvent;
import jakarta.ws.rs.sse.Sse;
import jakarta.ws.rs.sse.SseEventSink;

@Path("/events")
public class EventsResource {
    private final static ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);
    private final static AtomicInteger channelIds = new AtomicInteger();

    @GET
    @Produces(MediaType.SERVER_SENT_EVENTS)
    public void subscribe(@Context SseEventSink sink, @Context Sse sse) {
          final int channelId = channelIds.getAndIncrement();
          final Runnable sendMessage = () -> {
                final OutboundSseEvent event = sse.newEventBuilder()
                    .name("message")
                    .data(String.class, "Hello client " + channelId + "!")
                    .build();
                sink.send(event);
          };
          executor.scheduleAtFixedRate(sendMessage, 0, 1, TimeUnit.SECONDS);
    }
}

注册资源的Application类 #

我们在自定义的Application子类中注册上面的资源类：

import java.util.HashSet;
import java.util.Set;
import jakarta.ws.rs.core.Application;

public class RestApplication extends Application {
    /** {@inheritDoc} */
    @Override
    public Set<Class<?>> getClasses() {
        Set<Class<?>> classes = new HashSet<>();
        classes.add(EventsResource.class);
        return classes;
    }
}

启动Jetty的主类 #

最后是启动Jetty的主类，我们特意配置了一个只有5个线程的线程池来复现问题：

import org.eclipse.jetty.ee10.servlet.ServletContextHandler;
import org.eclipse.jetty.ee10.servlet.ServletHolder;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.example.rest.RestApplication;
import org.jboss.resteasy.plugins.server.servlet.HttpServletDispatcher;

public class Main {
  public static void main(String[] args) throws Exception {
    Server server = new Server(new QueuedThreadPool(5));

    ServerConnector serverConnector = new ServerConnector(server);
    serverConnector.setPort(8081);
    server.setConnectors(new Connector[] { serverConnector });

    ServletHolder servletHolder = new ServletHolder(new HttpServletDispatcher());
    servletHolder.setInitParameter("jakarta.ws.rs.Application", RestApplication.class.getName());
    servletHolder.setAsyncSupported(true);
    ServletContextHandler servletContextHandler = new ServletContextHandler(ServletContextHandler.SESSIONS);
    servletContextHandler.addServlet(servletHolder, "/*");
    server.setHandler(servletContextHandler);

    server.start();
    server.join();
  }
}

线程池耗尽的复现 #

当我们用curl订阅SSE时，可以正常收到事件：

% curl "http://localhost:8081/events" -H 'Accept: text/event-stream'

event: message
data: 0: Hello client 0!

event: message
data: 1: Hello client 0!

event: message
data: 2: Hello client 0!

...

但如果启动2个并行的curl请求，Jetty的5线程池就会被耗尽，此时第3个请求就会挂起，一直等待可用线程释放。

备注：内容来源于stack exchange，提问作者mgd