Использование функции неблокирующего ввода-вывода сервлетов 3.1 на tomcat 8

Question

Кто-нибудь пробовал сервлет 3.1 без блокировки на tomcat?

Запрос от браузера, кажется, ждет навсегда, но когда я запускаю сервер в режиме отладки, вызов возвращается, но я все еще не вижу «Data read..» и «Data written..» в журналах.

Servlet:

@WebServlet(urlPatterns = "/asyncn", asyncSupported = true) 
public class AsyncN extends HttpServlet { 

    private static final long serialVersionUID = 1L; 

    @Override 
    protected void service(final HttpServletRequest request, final HttpServletResponse response) throws ServletException, IOException { 

     println("Before starting job"); 

     final AsyncContext actx = request.startAsync(); 
     actx.setTimeout(Long.MAX_VALUE); 
     actx.start(new HeavyTask(actx)); 

     println("After starting job"); 

    } 

    class HeavyTask implements Runnable { 
     AsyncContext actx; 

     HeavyTask(AsyncContext actx) { 
      this.actx = actx; 
     } 

     @Override 
     public void run() { 
      try { 
       Thread.currentThread().setName("Job-Thread-" + actx.getRequest().getParameter("job")); 
       // set up ReadListener to read data for processing 
       ServletInputStream input = actx.getRequest().getInputStream(); 
       ReadListener readListener = new ReadListenerImpl(input, actx); 
       input.setReadListener(readListener); 
      } catch (IllegalStateException e) { 
       e.printStackTrace(); 
      } catch (IOException e) { 
       e.printStackTrace(); 
      } 
     } 
    } 

    public static void println(String output) { 
     System.out.println("[" + Thread.currentThread().getName() + "]" + output); 
    } 
} 

Слушателей:

public class ReadListenerImpl implements ReadListener { 

    private ServletInputStream input = null; 
    private AsyncContext actx = null; 

    // store the processed data to be sent back to client later 
    private Queue<String> queue = new LinkedBlockingQueue<>(); 

    ReadListenerImpl(ServletInputStream input, AsyncContext actx) { 
     this.input = input; 
     this.actx = actx; 
    } 

    @Override 
    public void onDataAvailable() throws IOException { 
     println("Data is now available, starting to read"); 
     StringBuilder sb = new StringBuilder(); 
     int len = -1; 
     byte b[] = new byte[8]; 
     // We need to check input#isReady before reading data. 
     // The ReadListener will be invoked again when 
     // the input#isReady is changed from false to true 
     while (input.isReady() && (len = input.read(b)) != -1) { 
      String data = new String(b, 0, len); 
      sb.append(data); 
     } 
     println("Data read: "+sb.toString()); 
     queue.add(sb.toString()); 
    } 

    @Override 
    public void onAllDataRead() throws IOException { 
     println("All Data read, now invoking write listener"); 
     // now all data are read, set up a WriteListener to write 
     ServletOutputStream output = actx.getResponse().getOutputStream(); 
     WriteListener writeListener = new WriteListenerImpl(output, queue, actx); 
     output.setWriteListener(writeListener); 
    } 

    @Override 
    public void onError(Throwable throwable) { 
     println("onError"); 
     actx.complete(); 
     throwable.printStackTrace(); 
    } 

    public static void println(String output) { 
     System.out.println("[" + Thread.currentThread().getName() + "]" + output); 
    } 
} 

public class WriteListenerImpl implements WriteListener { 

    private ServletOutputStream output = null; 
    private Queue<String> queue = null; 
    private AsyncContext actx = null; 

    WriteListenerImpl(ServletOutputStream output, Queue<String> queue, AsyncContext actx) { 
     this.output = output; 
     this.queue = queue; 
     this.actx = actx; 
    } 

    @Override 
    public void onWritePossible() throws IOException { 
     println("Ready to write, writing data"); 
     // write while there is data and is ready to write 
     while (queue.peek() != null && output.isReady()) { 
      String data = queue.poll(); 
      //do some processing here with the data 
      try { 
       data = data.toUpperCase(); 
       Thread.sleep(3000); 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } 
      println("Data written: "+data); 
      output.print(data); 
     } 
     // complete the async process when there is no more data to write 
     if (queue.peek() == null) { 
      actx.complete(); 
     } 
    } 

    @Override 
    public void onError(Throwable throwable) { 
     println("onError"); 
     actx.complete(); 
     throwable.printStackTrace(); 
    } 

    public static void println(String output) { 
     System.out.println("[" + Thread.currentThread().getName() + "]" + output); 
    } 
} 

SYSOUT пиловочник:

[http-nio-8080-exec-4]Before starting job 
[http-nio-8080-exec-4]After starting job 

журналы SYSOUT (при запуске сервера в режиме отладки):

[http-nio-8080-exec-6]Before starting job 
[http-nio-8080-exec-6]After starting job 
[http-nio-8080-exec-6]All Data read, now invoking write listener 
[http-nio-8080-exec-6]Ready to write, writing data 

Answer 1

Создание нового потока не требуется, установите readListener из метода службы, и все будет работать асинхронно.

Пара комментариев к вашему коду. В readListener у вас есть:

while (input.isReady() && (len = input.read(b)) != -1) 

бы предложил вместо использования этого полностью придерживаться асинхронной API:

while (input.isReady() && !input.isFinished()) 

Также для вашего слушателя записи у вас есть:

while (queue.peek() != null && output.isReady()) 

вы должны отменить условные обозначения до:

while (output.isReady() && queue.peek() != null) 

это защищает от вызова ac.complete() раньше, если самая последняя запись идет асинхронно.