淺談servlet3異步原理與實(shí)踐

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的超時(shí)時(shí)間,異步Servlet有對(duì)應(yīng)的超時(shí)時(shí)間，如果在指定的時(shí)間內(nèi)沒有執(zhí)行完操作，response依然會(huì)走原來Servlet的結(jié)束邏輯，后續(xù)的異步操作執(zhí)行完再寫回的時(shí)候，可能會(huì)遇到異常。

    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.
 * 在監(jiān)聽中初始化線程池
 */
@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<Runnable>(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.
 * 業(yè)務(wù)工作線程
 */
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<ResponseEntity<String>> testProcess(@PathVariable String testUrl) {
    final DeferredResult<ResponseEntity<String>> deferredResult = new DeferredResult<ResponseEntity<String>>();

    // 業(yè)務(wù)邏輯異步處理,將處理結(jié)果 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() {
      //業(yè)務(wù)邏輯START
      //...
      //業(yè)務(wù)邏輯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對(duì)象

    actx.setTimeout(30*3000);//設(shè)置異步調(diào)用超時(shí)時(shí)長

    ServletInputStream in = request.getInputStream();
    //異步讀?。▽?shí)現(xiàn)了非阻塞式讀取）
    in.setReadListener(new MyReadListener(in,actx));
    //直接輸出到頁面的內(nèi)容(不等異步完成就直接給頁面)
    PrintWriter out = response.getWriter();
    out.println("<h2>直接返回頁面，不等異步處理結(jié)果了</h2>");
    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;
  }
  //數(shù)據(jù)可用時(shí)觸發(fā)執(zhí)行
  @Override
  public void onDataAvailable() throws IOException {
    System.out.println("數(shù)據(jù)可用時(shí)觸發(fā)執(zhí)行");
  }

  //數(shù)據(jù)讀完時(shí)觸發(fā)調(diào)用
  @Override
  public void onAllDataRead() throws IOException {
    try {
      Thread.sleep(3000);//暫停5秒，模擬耗時(shí)處理數(shù)據(jù)
      PrintWriter out = asyncContext.getResponse().getWriter();
      out.write("數(shù)據(jù)讀完了");
      out.flush();
      System.out.println("數(shù)據(jù)讀完了");
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

  }

  //數(shù)據(jù)出錯(cuò)觸發(fā)調(diào)用
  @Override
  public void onError(Throwable t){
    System.out.println("數(shù)據(jù) 出錯(cuò)");
    t.printStackTrace();
  }
}