在等待“长”时间时重用tomcat线程

Question

配置
网络服务器：Nginx
应用服务器：Tomcat，默认配置为 200 个请求服务线程
我的服务器的预期响应时间：~30 秒（有很多第三方依赖项）

情景
应用程序需要每隔 10 秒生成一次令牌以供其使用。生成令牌的预期时间约为 5 秒，但由于它是通过网络联系的第三方系统，这显然是不一致的，可能长达 10 秒。
在令牌生成过程中，每秒将近 80% 的传入请求需要等待。

我认为应该发生的事情
由于等待令牌生成的请求将不得不等待“长时间”，因此在等待令牌生成过程完成时，没有理由将这些请求服务重新用于服务其他传入请求。
基本上，如果我的 20% 继续得到服务是有道理的。如果等待线程没有被用于其他请求，则将达到 tomcat 请求服务限制，服务器基本上会阻塞，这不是任何开发人员都会喜欢的。

我尝试了什么
最初我希望切换到 tomcat NIO 连接器可以完成这项工作。但是看了this的对比后，真的不抱希望了。不过，我尝试强制请求等待 10 秒，但没有成功。
现在我正在考虑我需要在等待时搁置请求并需要向 tomcat 发出该线程可以免费重用的信号。同样，当请求准备好向前移动时，我将需要 tomcat 从它的线程池中给我一个线程。但我对如何做到这一点甚至是否可能都一无所知。

任何指导或帮助？

Answer 1

您需要一个异步 servlet，但还需要对外部令牌生成器进行异步 HTTP 调用。如果您仍然在每个令牌请求的某处创建一个线程，则通过将请求从 servlet 传递到具有线程池的 ExecutorService 将一无所获。您必须将线程与 HTTP 请求分离，以便一个线程可以处理多个 HTTP 请求。这可以通过 Apache Asynch HttpClient 或 Async Http Client 等异步 HTTP 客户端来实现。

首先你必须创建一个像这样的异步 servlet

public class ProxyService extends HttpServlet {

    private CloseableHttpAsyncClient httpClient;

    @Override
    public void init() throws ServletException {
        httpClient = HttpAsyncClients.custom().
                setMaxConnTotal(Integer.parseInt(getInitParameter("maxtotalconnections"))).             
                setMaxConnPerRoute(Integer.parseInt(getInitParameter("maxconnectionsperroute"))).
                build();
        httpClient.start();
    }

    @Override
    public void destroy() {
        try {
            httpClient.close();
        } catch (IOException e) { }
    }

    @Override
    public void doGet(HttpServletRequest request, HttpServletResponse response) {
        AsyncContext asyncCtx = request.startAsync(request, response);
        asyncCtx.setTimeout(ExternalServiceMock.TIMEOUT_SECONDS * ExternalServiceMock.K);       
        ResponseListener listener = new ResponseListener();
        asyncCtx.addListener(listener);
        Future<String> result = httpClient.execute(HttpAsyncMethods.createGet(getInitParameter("serviceurl")), new ResponseConsumer(asyncCtx), null);
    }

}

此 servlet 使用 Apache Asynch HttpClient 执行异步 HTTP 调用。请注意，您可能希望配置每个路由的最大连接数，因为根据 RFC 2616 规范，HttpAsyncClient 默认只允许最多两个并发连接到同一主机。您还可以配置许多其他选项，如HttpAsyncClient configuration 所示。创建 HttpAsyncClient 的成本很高，因此您不想在每个 GET 操作上创建它的实例。

一个监听器挂接到AsyncContext，这个监听器只在上面的例子中用于处理超时。

public class ResponseListener implements AsyncListener {

    @Override
    public void onStartAsync(AsyncEvent event) throws IOException {
    }

    @Override
    public void onComplete(AsyncEvent event) throws IOException {
    }

    @Override
    public void onError(AsyncEvent event) throws IOException {
        event.getAsyncContext().getResponse().getWriter().print("error:");
    }

    @Override
    public void onTimeout(AsyncEvent event) throws IOException {
        event.getAsyncContext().getResponse().getWriter().print("timeout:");
    }

}

那么你需要一个 HTTP 客户端的消费者。当 HttpClient 在内部执行 buildResult() 作为向调用者 ProxyService servlet 返回 Future<String> 的步骤时，此使用者通过调用 complete() 来通知 AsyncContext。

public class ResponseConsumer extends AsyncCharConsumer<String> {

    private int responseCode;
    private StringBuilder responseBuffer;
    private AsyncContext asyncCtx;

    public ResponseConsumer(AsyncContext asyncCtx) {
        this.responseBuffer = new StringBuilder();
        this.asyncCtx = asyncCtx;
    }

    @Override
    protected void releaseResources() { }

    @Override
    protected String buildResult(final HttpContext context) {
        try {
            PrintWriter responseWriter = asyncCtx.getResponse().getWriter();
            switch (responseCode) {
                case javax.servlet.http.HttpServletResponse.SC_OK:
                    responseWriter.print("success:" + responseBuffer.toString());
                    break;
                default:
                    responseWriter.print("error:" + responseBuffer.toString());
                }
        } catch (IOException e) { }
        asyncCtx.complete();        
        return responseBuffer.toString();
    }

    @Override
    protected void onCharReceived(CharBuffer buffer, IOControl ioc) throws IOException {
        while (buffer.hasRemaining())
            responseBuffer.append(buffer.get());
    }

    @Override
    protected void onResponseReceived(HttpResponse response) throws HttpException, IOException {        
        responseCode = response.getStatusLine().getStatusCode();
    }

}

ProxyService servlet 的 web.xml 配置可能是这样的

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:web="http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
         id="WebApp_ID" version="3.0" metadata-complete="true">
  <display-name>asyncservlet-demo</display-name>

  <servlet>
    <servlet-name>External Service Mock</servlet-name>
    <servlet-class>ExternalServiceMock</servlet-class>
    <load-on-startup>1</load-on-startup>
  </servlet>

  <servlet>
    <servlet-name>Proxy Service</servlet-name>
    <servlet-class>ProxyService</servlet-class>
    <load-on-startup>1</load-on-startup>
    <async-supported>true</async-supported>
    <init-param>
      <param-name>maxtotalconnections</param-name>
      <param-value>200</param-value>
    </init-param>
    <init-param>
      <param-name>maxconnectionsperroute</param-name>
      <param-value>4</param-value>
    </init-param>
    <init-param>
      <param-name>serviceurl</param-name>
      <param-value>http://127.0.0.1:8080/asyncservlet/externalservicemock</param-value>
    </init-param>
  </servlet>

  <servlet-mapping>
    <servlet-name>External Service Mock</servlet-name>
    <url-pattern>/externalservicemock</url-pattern>
  </servlet-mapping>

  <servlet-mapping>
    <servlet-name>Proxy Service</servlet-name>
    <url-pattern>/proxyservice</url-pattern>
  </servlet-mapping>

</web-app>

令牌生成器的模拟 servlet 可能会延迟几秒：

public class ExternalServiceMock extends HttpServlet{

    public static final int TIMEOUT_SECONDS = 13;
    public static final long K = 1000l;

    public void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException {
        Random rnd = new Random();
        try {
            Thread.sleep(rnd.nextInt(TIMEOUT_SECONDS) * K);
        } catch (InterruptedException e) { }
        final byte[] token = String.format("%10d", Math.abs(rnd.nextLong())).getBytes(ISO_8859_1);
        response.setContentType("text/plain");
        response.setCharacterEncoding(ISO_8859_1.name());
        response.setContentLength(token.length);
        response.getOutputStream().write(token);
    }

}

你可以得到一个fully working example at GitHub。

Answer 2

这个问题本质上是存在这么多“反应式”库和工具包的原因。

这不是一个可以通过调整或更换 tomcat 连接器来解决的问题。
您基本上需要删除所有阻塞 IO 调用，用非阻塞 IO 替换它们可能需要重写大部分应用程序。
您的 HTTP 服务器需要是非阻塞的，您需要对服务器使用非阻塞 API（如 servlet 3.1），并且您对第三方 API 的调用必须是非阻塞的。
Vert.x 和 RxJava 等库提供了工具来帮助解决所有这些问题。

否则唯一的其他选择就是增加线程池的大小，操作系统已经负责调度 CPU 以便不活动的线程不会造成太多的性能损失，但总会有更多与反应式方法相比的开销。

如果不了解您的应用程序的更多信息，就很难就具体方法提供建议。

Answer 3

使用异步 servlet 请求或响应式库（如其他答案中所述）会有所帮助，但需要进行重大的架构更改。

另一种选择是将令牌更新与令牌使用分开。

这是一个简单的实现：

public class TokenHolder {
    public static volatile Token token = null;
    private static Timer timer = new Timer(true);
    static {
        // set the token 1st time
        TokenHolder.token = getNewToken();

        // schedule updates periodically
        timer.schedule(new TimerTask(){
            public void run() {
                TokenHolder.token = getNewToken();
            }
        }, 10000, 10000);
    }
}

现在您的请求只需使用TokenHolder.token 即可访问该服务。

在实际应用中，您可能会使用更高级的调度工具。