java - 在 tomcat 8 上使用 Servlets 3.1 的非阻塞 I/O 特性

Question

有没有人在tomcat上试过Servlets 3.1非阻塞技术？

来自浏览器的请求似乎永远在等待，但是当我在 Debug模式下运行服务器时，调用返回但我仍然没有看到“Data read..”和“数据写入..”在日志中。

小服务程序:

@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);
    }
}

系统输出日志:

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

Sysout 日志(当我在 Debug模式下运行服务器时):

[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()。