等待"漫长"时间重用tomcat线程

Question

配置
Web服务器:Nginx
App服务器:Tomcat,默认配置为200个请求服务线程
我服务器的预期响应时间:~30秒(有很多第三方依赖项)

情景
每10秒钟,应用程序需要生成令牌以供其使用.令牌生成的预期时间约为5秒,但由于其第三方系统通过网络进行联系,因此显然不一致,最长可达10秒.
在令牌生成过程中,每秒几乎80%的传入请求都需要等待.

我认为应该发生的事情
由于等待令牌生成的请求必须等待"很长"的时间,因此没有理由在等待令牌生成过程完成时重新使用这些请求服务来服务其他传入请求.
基本上,如果我的20%继续服务是有意义的.如果等待的线程没有被用于其他请求,那么将达到tomcat请求服务限制,服务器将基本上阻塞,而不是任何开发人员想要的东西.

我做了什么
最初我希望切换到tomcat NIO连接器可以完成这项工作.但看了这个比较后,我真的没有希望.尽管如此,我试图迫使请求等待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配置中所示.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客户端的使用者.此消费者通过调用complete()何时在buildResult()HttpClient内部执行作为返回Future<String>调用者ProxyServiceservlet 的步骤来通知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);
    }

}

你可以在GitHub上找到一个完整的例子.