浅谈servlet3异步原理与实践

package com.test.servlet3;

import javax.servlet.AsyncContext;
import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * Created by wangxindong on 2017/10/19.
 */
@WebServlet(urlPatterns = "/AsyncLongRunningServlet", asyncSupported = true)
public class AsyncLongRunningServlet extends HttpServlet {
  private static final long serialVersionUID = 1L;

  protected void doGet(HttpServletRequest request,
             HttpServletResponse response) throws ServletException, IOException {
    long startTime = System.currentTimeMillis();
    System.out.println("AsyncLongRunningServlet Start::Name="
        + Thread.currentThread().getName() + "::ID="
        + Thread.currentThread().getId());

    request.setAttribute("org.apache.catalina.ASYNC_SUPPORTED", true);

    String time = request.getParameter("time");
    int secs = Integer.valueOf(time);
    // max 10 seconds
    if (secs > 10000)
      secs = 10000;

    AsyncContext asyncCtx = request.startAsync();
    asyncCtx.addListener(new AppAsyncListener());
    asyncCtx.setTimeout(9000);//异步servlet的超时时间,异步Servlet有对应的超时时间，如果在指定的时间内没有执行完操作，response依然会走原来Servlet的结束逻辑，后续的异步操作执行完再写回的时候，可能会遇到异常。

    ThreadPoolExecutor executor = (ThreadPoolExecutor) request
        .getServletContext().getAttribute("executor");

    executor.execute(new AsyncRequestProcessor(asyncCtx, secs));
    long endTime = System.currentTimeMillis();
    System.out.println("AsyncLongRunningServlet End::Name="
        + Thread.currentThread().getName() + "::ID="
        + Thread.currentThread().getId() + "::Time Taken="
        + (endTime - startTime) + " ms.");
  }
}

package com.test.servlet3;

import javax.servlet.AsyncEvent;
import javax.servlet.AsyncListener;
import javax.servlet.ServletResponse;
import javax.servlet.annotation.WebListener;
import java.io.IOException;
import java.io.PrintWriter;

/**
 * Created by wangxindong on 2017/10/19.
 */
@WebListener
public class AppAsyncListener implements AsyncListener {
  @Override
  public void onComplete(AsyncEvent asyncEvent) throws IOException {
    System.out.println("AppAsyncListener onComplete");
    // we can do resource cleanup activity here
  }

  @Override
  public void onError(AsyncEvent asyncEvent) throws IOException {
    System.out.println("AppAsyncListener onError");
    //we can return error response to client
  }

  @Override
  public void onStartAsync(AsyncEvent asyncEvent) throws IOException {
    System.out.println("AppAsyncListener onStartAsync");
    //we can log the event here
  }

  @Override
  public void onTimeout(AsyncEvent asyncEvent) throws IOException {
    System.out.println("AppAsyncListener onTimeout");
    //we can send appropriate response to client
    ServletResponse response = asyncEvent.getAsyncContext().getResponse();
    PrintWriter out = response.getWriter();
    out.write("TimeOut Error in Processing");
  }
}

package com.test.servlet3;

import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import javax.servlet.annotation.WebListener;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by wangxindong on 2017/10/19.
 * 在监听中初始化线程池
 */
@WebListener
public class AppContextListener implements ServletContextListener {
  public void contextInitialized(ServletContextEvent servletContextEvent) {

    // create the thread pool
    ThreadPoolExecutor executor = new ThreadPoolExecutor(100, 200, 50000L,
        TimeUnit.MILLISECONDS, new ArrayBlockingQueue(100));
    servletContextEvent.getServletContext().setAttribute("executor",
        executor);

  }

  public void contextDestroyed(ServletContextEvent servletContextEvent) {
    ThreadPoolExecutor executor = (ThreadPoolExecutor) servletContextEvent
        .getServletContext().getAttribute("executor");
    executor.shutdown();
  }
}

package com.test.servlet3;

import javax.servlet.AsyncContext;
import java.io.IOException;
import java.io.PrintWriter;

/**
 * Created by wangxindong on 2017/10/19.
 * 业务工作线程
 */
public class AsyncRequestProcessor implements Runnable {
  private AsyncContext asyncContext;
  private int secs;

  public AsyncRequestProcessor() {
  }

  public AsyncRequestProcessor(AsyncContext asyncCtx, int secs) {
    this.asyncContext = asyncCtx;
    this.secs = secs;
  }

  @Override
  public void run() {
    System.out.println("Async Supported? "
        + asyncContext.getRequest().isAsyncSupported());
    longProcessing(secs);
    try {
      PrintWriter out = asyncContext.getResponse().getWriter();
      out.write("Processing done for " + secs + " milliseconds!!");
    } catch (IOException e) {
      e.printStackTrace();
    }
    //complete the processing
    asyncContext.complete();
  }

  private void longProcessing(int secs) {
    // wait for given time before finishing
    try {
      Thread.sleep(secs);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

@Controller
@RequestMapping("/async/TestController")
public class TestController {
  @ResponseBody
  @RequestMapping("/{testUrl}")
  public DeferredResult> testProcess(@PathVariable String testUrl) {
    final DeferredResult> deferredResult = new DeferredResult>();

    // 业务逻辑异步处理,将处理结果 set 到 DeferredResult
    new Thread(new AsyncTask(deferredResult)).start();

    return deferredResult;
  }

  private static class AsyncTask implements Runnable {

    private DeferredResult result;

    private AsyncTask(DeferredResult result) {
      this.result = result;
    }

    @Override
    public void run() {
      //业务逻辑START
      //...
      //业务逻辑END
      result.setResult(result);
    }
  }
}

package com.test.servlet3Noblock;

import javax.servlet.AsyncContext;
import javax.servlet.ServletException;
import javax.servlet.ServletInputStream;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.PrintWriter;

/**
 * Created by wangxindong on 2017/10/23.
 */
@WebServlet(urlPatterns = "/AsyncLongRunningServlet2", asyncSupported = true)
public class AsyncLongRunningServlet extends HttpServlet {
  protected void doGet(HttpServletRequest request,
             HttpServletResponse response) throws ServletException, IOException {
    request.setCharacterEncoding("UTF-8");
    response.setContentType("text/html;charset=UTF-8");

    AsyncContext actx = request.startAsync();//通过request获得AsyncContent对象

    actx.setTimeout(30*3000);//设置异步调用超时时长

    ServletInputStream in = request.getInputStream();
    //异步读取（实现了非阻塞式读取）
    in.setReadListener(new MyReadListener(in,actx));
    //直接输出到页面的内容(不等异步完成就直接给页面)
    PrintWriter out = response.getWriter();
    out.println("
直接返回页面，不等异步处理结果了
");
    out.flush();
  }

}

package com.test.servlet3Noblock;

import javax.servlet.AsyncContext;
import javax.servlet.ReadListener;
import javax.servlet.ServletInputStream;
import java.io.IOException;
import java.io.PrintWriter;

/**
 * Created by wangxindong on 2017/10/23.
 */
public class MyReadListener implements ReadListener {
  private ServletInputStream inputStream;
  private AsyncContext asyncContext;
  public MyReadListener(ServletInputStream input,AsyncContext context){
    this.inputStream = input;
    this.asyncContext = context;
  }
  //数据可用时触发执行
  @Override
  public void onDataAvailable() throws IOException {
    System.out.println("数据可用时触发执行");
  }

  //数据读完时触发调用
  @Override
  public void onAllDataRead() throws IOException {
    try {
      Thread.sleep(3000);//暂停5秒，模拟耗时处理数据
      PrintWriter out = asyncContext.getResponse().getWriter();
      out.write("数据读完了");
      out.flush();
      System.out.println("数据读完了");
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

  }

  //数据出错触发调用
  @Override
  public void onError(Throwable t){
    System.out.println("数据 出错");
    t.printStackTrace();
  }
}