在JavaWeb应用中对客户请求的异步处理

Posted 2022-03-31

本文结合具体的范例，介绍如何在JavaWeb应用中对客户请求进行异步处理，在Servlet中进行文件上传。本文的参考书籍是《Tomcat与Java Web开发技术详解》第三版，作者：孙卫琴。

本文所用的软件版本为：Window10，JDK10，Tomcat9。

本文所涉及的源代码的下载网址为：
http://www.javathinker.net/javaweb/upload-app.rar

在Servlet API 3.0版本之前，Servlet容器针对每个HTTP请求都会分配一个工作线程。即对于每一次HTTP请求，Servlet容器都会从主线程池中取出一个空闲的工作线程，由该线程从头到尾负责处理请求。如果在响应某个HTTP请求的过程中涉及到进行I/O操作、访问数据库，或其他耗时的操作，那么该工作线程会被长时间占用，只有当工作线程完成了对当前HTTP请求的响应，才能释放回线程池以供后续使用。
在并发访问量很大的情况下，如果线程池中的许多工作线程都被长时间占用，这将严重影响服务器的并发访问性能。所谓并发访问性能，是指服务器在同一时间可以同时响应众多客户请求的能力。为了解决这种问题，从Servlet API 3.0版本开始，引入了异步处理机制，随后在Servlet API 3.1中又引入了非阻塞I/O来进一步增强异步处理的性能。
Servlet异步处理的机制为：Servlet从HttpServletRequest对象中获得一个AsyncContext对象，该对象表示异步处理的上下文。AsyncContext把响应当前请求的任务传给一个新的线程，由这个新的线程来完成对请求的处理并向客户端返回响应结果。最初由Servlet容器为HTTP请求分配的工作线程便可以及时地释放回主线程池，从而及时处理更多的请求。由此可以看出，所谓Servlet异步处理机制，就是把响应请求的任务从一个线程传给另一个线程来处理。
1.1 异步处理的流程
要创建支持异步处理的Serlvet类主要包含以下步骤：
（1）在Servlet类中把@WebServlet标注的asyncSupport属性设为true，使得该Servlet支持异步处理。例如：

@WebServlet(name="AsyncServlet1",
            urlPatterns="/async1",
            asyncSupported=true)

如果在web.xml文件中配置该Servlet，那么需要把<async-supported>元素设为true：

 <servlet>   
    <servlet-name>AsyncServlet1</servlet-name>   
    <servlet-class>mypack.AsyncServlet1</servlet-class>   
    <async-supported>true</async-supported>   
  </servlet>

（2）在Servlet类的服务方法中，通过ServletRequest对象的startAsync()方法，获得AsyncContext对象：

AsyncContext asyncContext = request.startAsync();

AsyncContext接口为异步处理当前请求提供了上下文，它具有以下方法：
? setTimeout(long timeout)：设置异步线程处理请求任务的超时时间（以毫秒为单位），即异步线程必须在timeout参数指定的时间内完成任务。
? start(java.lang.Runnable run) ：启动一个异步线程，执行参数run指定的任务。
? addListener(AsyncListener listener) ：添加一个异步监听器。
? complete()：告诉Servlet容器任务完成，返回响应结果。
? dispatch(java.lang.String path) ：把请求派发給参数path指定的Web组件。
? getRequest() ：获得当前上下文中的ServletRequest对象。
? getResponse()：获得当前上下文中的ServletResponse对象。
（3）调用AsyncContext对象的setTimeout(long timeout) 设置异步线程的超时时间，这一步不是必须的。
（4）启动一个异步线程来执行处理请求的任务。关于如何启动异步线程，有三种方式，参见5.10.2节的例程5-25（AsyncServlet1.java）、例程5-27（AsyncServlet2.java）和例程5-28（AsyncServlet3.java）。
（5）调用AsyncContext对象的complete()方法来告诉Servlet容器已经完成任务，或者调用AsyncContext对象的的dispatch()方法把请求派发給其他Web组件。
1.2 异步处理的范例
以下例程1-1的AsyncServlet1类是一个支持异步处理的Servlet范例。
例程1-1 AsyncServlet1.java

package mypack;
import javax.servlet.*;
import javax.servlet.http.*;
import javax.servlet.annotation.*;
import java.io.*;

@WebServlet(name="AsyncServlet1",
            urlPatterns="/async1",
            asyncSupported=true)

public class AsyncServlet1 extends HttpServlet

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

     response.setContentType("text/plain;charset=GBK");
     AsyncContext asyncContext = request.startAsync();
     //设定异步操作的超时时间
     asyncContext.setTimeout(60*1000);  

     //启动异步线程的方式一
     asyncContext.start(new MyTask(asyncContext));
  

以上AsyncServlet1通过AsyncContext对象的start()方法来启动异步线程：
asyncContext.start(new MyTask(asyncContext));
异步线程启动后，就会执行MyTask对象的run()方法中的代码。AsyncContext接口的start()方法的实现方式取决于具体的Servlet容器。有的Servlet容器除了拥有存放工作线程的主线程池，还会另外维护一个线程池，从该线程池中取出空闲的线程来异步处理请求。
有的Servlet容器从已有的主线程池中获得一个空闲的线程来作为异步处理请求的线程，这种实现方式对性能的改进不大，因为如果异步线程和初始线程共享同一个线程池的话，就相当于先闲置初始工作线程，再占用另一个空闲的工作线程。
以下例程1-2的MyTask类定义了处理请求的具体任务，它实现了Runnable接口。
例程1-2 MyTask.java

package mypack;
import javax.servlet.*;
import javax.servlet.http.*;

public class MyTask implements Runnable
  private AsyncContext asyncContext;

  public MyTask(AsyncContext asyncContext)
    this.asyncContext = asyncContext;
  

  public void run()
    try
      //睡眠5秒，模拟很耗时的一段业务操作
      Thread.sleep(5*1000);
      asyncContext.getResponse()
                  .getWriter()
                  .write("让您久等了!");   
      asyncContext.complete();
    catch(Exception e)e.printStackTrace();
  

MyTask类利用AsyncContext对象的getResponse()方法来获得当前的ServletResponse对象，利用AsyncContext对象的complete()方法来通知Servlet容易已经完成任务。
通过浏览器访问：http://localhost:8080/helloapp/async1，会看到客户端在耐心等待了5秒钟后才会得到如下图1-1所示的响应结果。

图1-1 AsyncServlet1的响应结果

以下例程1-3的AsyncServlet2类介绍了启动异步线程的第二种方式。
例程1-3 AsyncServlet2.java

@WebServlet(name="AsyncServlet2",
            urlPatterns="/async2",
            asyncSupported=true)

public class AsyncServlet2 extends HttpServlet

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

     response.setContentType("text/plain;charset=GBK");
     AsyncContext asyncContext = request.startAsync();
     //设定异步操作的超时时间
     asyncContext.setTimeout(60*1000);  

     //启动异步线程的方式二
     new Thread(new MyTask(asyncContext)).start();
  

以上AsyncServlet2类通过“new Thread()”语句亲自创建新的线程，把它作为异步线程。当大量用户并发访问AsyncServlet2类时，会导致服务器端创建大量的新线程，这会大大降低服务器的运行性能。
以下例程1-4的AsyncServlet3类介绍了启动异步线程的第三种方式。
例程1-4 AsyncServlet3.java

package mypack;
import javax.servlet.*;
import javax.servlet.http.*;
import javax.servlet.annotation.*;
import java.io.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@WebServlet(name="AsyncServlet3",
            urlPatterns="/async3",
            asyncSupported=true)

public class AsyncServlet3 extends HttpServlet
  private static ThreadPoolExecutor executor = 
          new ThreadPoolExecutor(100, 200, 50000L, 
                 TimeUnit.MILLISECONDS, 
                 new ArrayBlockingQueue<>(100));

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

     response.setContentType("text/plain;charset=GBK");
     AsyncContext asyncContext = request.startAsync();
     //设定异步操作的超时时间
     asyncContext.setTimeout(60*1000); 

     //启动异步线程的方式三 
     executor.execute(new MyTask(asyncContext));
  

  public void destroy()
    //关闭线程池
    executor.shutdownNow();
  

以上AsyncServlet3类利用Java API中的线程池ThreadPoolExecutor类来创建一个线程池，所有的异步线程都存放在这个线程池中。图1-2演示了主线程池和异步处理线程池的关系。

图1-2 主线程池和异步处理线程池的关系

使用ThreadPoolExecutor线程池类的优点是可以更加灵活地根据实际应用需求来设置线程池。在构造ThreadPoolExecutor对象时就可以对线程池的各种选项进行设置。以下是ThreadPoolExecutor类的一个构造方法：

public ThreadPoolExecutor(int corePoolSize,
                     int maximumPoolSize,
                     long keepAliveTime,
                     TimeUnit unit,
                     BlockingQueue<Runnable> workQueue)

以上ThreadPoolExecutor类的构造方法包含以下参数：
? corePoolSize：线程池维护的线程的最少数量。
? maximumPoolSize：线程池维护的线程的最大数量。
? keepAliveTime：线程池维护的线程所允许的空闲时间。
? unit：线程池维护的线程所允许的空闲时间的单位。
? workQueue：线程池所使用的缓冲队列。

ThreadPoolExecutor类的execute(Runnable r)方法会从线程池中取出一个空闲的线程，来执行参数指定的任务：

executor.execute(new MyTask(asyncContext));

1.3 异步监听器
在异步处理请求的过程中，还可以利用异步监听器AsyncListener来捕获并处理异步线程运行中的特定事件。AsyncListener接口声明了四个方法：
? onStartAsync(AsyncEvent event)：异步线程开始时调用。
? onError(AsyncEvent event)： 异步线程出错时调用。
? onTimeout(AsyncEvent event)： 异步线程执行超时时调用。
? onComplete(AsyncEvent event)： 异步线程执行完毕时调用。
以下例程1-5的AsyncServlet4与1.2节的例程1-1的AsyncServlet1类很相似。区别在于AsyncServlet4类中的AsyncContext对象注册了AsyncListener监听器。
例程1-5 AsyncServlet4.java

@WebServlet(name="AsyncServlet4",
            urlPatterns="/async4",
            asyncSupported=true)

public class AsyncServlet4 extends HttpServlet
  public void service(HttpServletRequest request,
              HttpServletResponse response)
              throws ServletException,IOException  

     response.setContentType("text/plain;charset=GBK");
     AsyncContext asyncContext = request.startAsync();
     //设定异步操作的超时时间
     asyncContext.setTimeout(60*1000); 

     //注册异步处理监听器 
     asyncContext.addListener(new AsyncListener()

       public void onComplete(AsyncEvent asyncEvent) 
                                throws IOException
         System.out.println("on Complete...");
       

       public void onTimeout(AsyncEvent asyncEvent) 
                                throws IOException
         System.out.println("on Timeout...");
       

       public void onError(AsyncEvent asyncEvent)
                                throws IOException
         System.out.println("on Error...");
       

       public void onStartAsync(AsyncEvent asyncEvent)
                                throws IOException
         System.out.println("on Start...");
       
     );

     asyncContext.start(new MyTask(asyncContext));
  

以上AsyncContext对象所注册的异步监听器是一个内部匿名类，它实现了AsyncListener接口的各个方法，能够在异步线程启动、出错、超时或结束时在服务器的控制台打印出特定的语句。
1.4 非阻塞I/O
非阻塞I/O是与阻塞I/O相对的概念。阻塞I/O包括以下两种情况：
? 当一个线程在通过输入流执行读操作时，如果输入流的可读数据暂时还未准备好，那么当前线程会进入阻塞状态（也可理解为等待状态），只有当读到了数据或者到达了数据末尾，线程才会从读方法中退出。例如服务器端读取客户端发送的请求数据时，如果请求数据很大（比如上传文件），那么这些数据在网络上传输需要耗费一些时间，此时服务器端负责读取请求数据的线程可能会进入阻塞状态。
? 当一个线程在通过输出流执行写操作时，如果因为某种原因，暂时不能向目的地写数据，那么当前线程会进入阻塞状态，只有当完成了写数据的操作，线程才会从写方法中退出。例如当服务器端向客户端发送响应结果时，如果响应正文很大（比如下载文件），那么这些数据在网络上传输需要耗费一些时间，此时服务器端负责输出响应结果的线程可能会进入阻塞状态。
非阻塞I/O操作也包括两种情况：
? 当一个线程在通过输入流执行读操作时，如果输入流的可读数据暂时还未准备好，那么当前线程不会进入阻塞状态，而是立即退出读方法。只有当输入流中有可读数据时，再进行读操作。
? 当一个线程在通过输出流执行写操作时，如果因为某种原因，暂时不能向目的地写数据，那么当前线程不会进入阻塞状态，而是立即退出写方法。只有当可以向目的地写数据时，再进行写操作。
在Java语言中，传统的输入/输出操作都采用阻塞I/O的方式。本章前面几节已经介绍了如何用异步处理机制来提高服务器的并发访问性能。但是，当异步线程用阻塞I/O的方式来读写数据时，毕竟还是会使得异步线程常常进入阻塞状态，这还是会削弱服务器的并发访问性能。
为了解决上述问题，从Servlet API 3.1开始，引入了非阻塞I/O机制，它建立在异步处理的基础上，具体实现方式是引入了两个监听器：
? ReadListener接口：监听ServletInputStream输入流的行为。
? WriteListener接口：监听ServletOutputStream输出流的行为。
ReadListener接口包含以下方法：
? onDataAvailable()：输入流中有可读数据时触发此方法。
? onAllDataRead()：输入流中所有数据读完时触发此方法。
? onError(Throwable t)：输入操作出现错误时触发此方法。
WriteListener接口包含以下方法：
? onWritePossible()：可以向输出流写数据时触发此方法。
? onError(java.lang.Throwable throwable)：输出操作出现错误时触发此方法。
在支持异步处理的Servlet类中进行非阻塞I/O操作主要包括以下步骤：
（1）在服务方法中从ServletRequest对象或ServletResponse对象中得到输入流或输出流：
ServletInputStream input = request.getInputStream();
ServletOutputStream output = request.getOutputStream();
（2）为输入流注册一个读监听器，或为输出流注册一个写监听器：
//以下context引用AsyncContext对象
input.setReadListener(new MyReadListener(input, context));
output.setWriteListener(new MyWriteListener(output, context));
（3）在读监听器类或写监听器类中编写包含非阻塞I/O操作的代码 。
下面通过具体范例来演示非阻塞I/O的用法。本范例涉及到三个Web组件：upload2.htm?NoblockServlet.java?OutputServlet.java。
upload2.htm会生成一个可以上传文件的网页，它的主要源代码如下：

  <form name="uploadForm" method="POST"
    enctype="MULTIPART/FORM-DATA"
    action="nonblock">
    <table>
      <tr>
       <td><div align="right">User Name:</div></td>
       <td><input type="text" name="username" size="30"/> </td>
      </tr>
      <tr>
       <td><div align="right">Upload File1:</div></td>
       <td><input type="file" name="file1" size="30"/> </td>
      </tr>
      <tr>
        <td><input type="submit" name="submit" value="upload"></td>
        <td><input type="reset" name="reset" value="reset"></td>
      </tr>
    </table>
  </form>
OutputServlet.java的作用是向网页上输出请求范围内的msg属性的值，以下是它的源代码：
public class OutputServlet extends GenericServlet 

  public void service(ServletRequest request,
              ServletResponse response)
              throws ServletException, IOException 

    //读取CheckServlet存放在请求范围内的消息
    String message = (String)request.getAttribute("msg");
    PrintWriter out=response.getWriter();

    out.println(message); 
    out.close();
  

以下例程1-6是NonblockServlet类的源代码，它为ServletInputStream注册了读监听器，并且在service()方法的开头和结尾，会向客户端打印进入service()方法以及退出service()方法的时间。
例程1-6 NonblockServlet.java

package mypack;
import javax.servlet.*;
import javax.servlet.http.*;
import javax.servlet.annotation.*;
import java.io.*;

@WebServlet(urlPatterns="/nonblock",
            asyncSupported=true)
public class NonblockServlet extends HttpServlet

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

    response.setContentType("text/html;charset=GBK");
    PrintWriter out = response.getWriter();
    out.println("<title>非阻塞IO示例</title>");
    out.println("进入Servlet的service()方法的时间："
      + new java.util.Date() + ".<br/>");

    // 创建AsyncContext 
    AsyncContext context = request.startAsync();
    //设置异步调用的超时时长
    context.setTimeout(60 * 1000);

    ServletInputStream input = request.getInputStream();
    //为输入流注册监听器
    input.setReadListener(new MyReadListener(input, context));

    out.println("退出Servlet的service()方法的时间："
               + new java.util.Date() + ".<br/><hr>");
    out.flush();
  

以上ServletInputStream注册的读监听器为MyReadListener类，以下例程1-7是它的源代码。
例程1-7 MyReadListener.java

package mypack;
import javax.servlet.*;
import javax.servlet.http.*;
import java.io.*;

public class MyReadListener implements ReadListener
  private ServletInputStream input;
  private AsyncContext context;
  private StringBuilder sb = new StringBuilder();

  public MyReadListener(ServletInputStream input , 
                        AsyncContext context)
    this.input = input;
    this.context = context;
  

  public void onDataAvailable()
    System.out.println("数据可用！");
    try
      // 暂停5秒，模拟读取数据是一个耗时操作。
      Thread.sleep(5000);

       int len = -1;
      byte[] buff = new byte[1024];

      //读取浏览器向Servlet提交的数据
      while (input.isReady() && (len = input.read(buff)) > 0)
        String data = new String(buff , 0 , len);
        sb.append(data);
      
    catch (Exception ex)ex.printStackTrace();
  

  public void onAllDataRead()
    System.out.println("数据读取完成！");
    System.out.println(sb);
    //将数据设置为request范围的属性
    context.getRequest().setAttribute("msg" , sb.toString());
    //把请求派发給OutputServlet组件
    context.dispatch("/output");
  

  public void onError(Throwable t)
    t.printStackTrace();  
  

MyReadListener类实现了ReadListener接口中的所有方法。在onDataAvailable()方法中读取客户端的请求数据，把它存放到StringBuilder对象中。在onAllDataRead()方法中，把StringBuilder对象包含的字符串作为msg属性存放到请求范围内。最后把请求派发給URL为“/output”的Web组件来处理，它和OutputServlet对应。
通过浏览器访问http://localhost:8080/helloapp/upload2.htm，将会出现如图1-3所示的网页。

图1-3 upload2.htm网页

在网页中输入相关数据，再提交表单，该请求由URL为“/nonblock”的Web组件来处理，它和NonblockServlet组件对应。而NonblockServlet组件会通过MyReadListener读监听器采取非阻塞I/O的方式来读取请求数据，最后MyReadListener读监听器把请求派发給OutputServlet。NonblockServlet和OutputServlet共同生成的响应结果参见图1-4。

图1-4 NonblockServlet和OutputServlet共同生成的响应结果

在客户端等待图1-4的网页的内容全部展示出来的过程中，可以看出，当主工作线程已经退出NonblockServlet的service()方法时，读取客户请求数据的非阻塞I/O操作还没有完成。那么到底是由哪个线程来执行非阻塞I/O操作的呢？这取决于Servlet容器的实现，用户无需了解其中的细节，反正可以肯定的是，Servlet容器会提供一个异步线程来执行MyReadListener读监听器中的非阻塞I/O操作。