了解Jetty的“关闭时挂起/未读”警告

有可能复制 Closed while Pending/Unready 使用一些手动调试和简单的 curl

1. The first breakpoint is in the HttpOutput.write() method right after it changes its state from READY to PENDING :

   case READY:
       if (!_state.compareAndSet(OutputState.READY, OutputState.PENDING))
           continue;
       // put a breakpoint here
   

2. The second one in Response.closeOutput()

   case STREAM:
       // put a breakpiont here
       if (!_out.isClosed())
           getOutputStream().close();
       break;
   

复制步骤:

1. QueueWriteListener.onWritePossible() 它向输出写入几个字节,然后返回(因为它的输入缓冲区是空的)。
2. 等待onTimeout事件。
3. [喷气式飞机2] HttpChannelState.onTimeout() 电话 QueueWriteListener.onTimeout() 打电话来 asyncContext.complete() .
4. [喷气式飞机2] 在异步超时后计划调度。
5. [JettyThread2]在第二个断点处暂停
6. [干膜厚度] DataFeederThread 电话 writeImpl()
7. 数据馈送线程 电话 (这是输出流)
8. [干膜厚度] HttpOutput.write() 准备好了 到 悬而未决的
9. [干膜厚度] 由于上面的断点,此处暂停
10. 挂起/未读时关闭 警告。

所以,实际上是Jetty关闭了这个堆栈(Jetty 9.4.8.v20171121)上的输出流:

Thread [jetty-19] (Suspended)   
    Response.closeOutput() line: 1043   
    HttpChannelOverHttp(HttpChannel).handle() line: 488 
    HttpChannelOverHttp(HttpChannel).run() line: 293    
    QueuedThreadPool.runJob(Runnable) line: 708 
    QueuedThreadPool$2.run() line: 626  
    Thread.run() line: 748  

制作 onTimeout() synchronized (以及 写入impl() 也是 )在侦听器中没有帮助,因为计划的关闭仍然可以与 writeImpl 数据馈送线程 ). 考虑这个案例:

1. [JettyThread1]调用 它向输出写入几个字节,然后返回(因为它的输入缓冲区是空的)。
2. 等待onTimeout事件。
3. [喷气式飞机2] 电话 QueueWriteListener.onTimeout() 打电话来 asyncContext.complete()
4. [干膜厚度] 电话 写入impl() onTimeout 尚未完成)
5. [喷气式飞机2] QueueWriteListener.onTimeout()
6. [干膜厚度] 能跑
7. [喷气式飞机2] 在异步超时后计划调度。
8. [JettyThread2]在第二个断点处暂停
9. 数据馈送线程 电话 HttpOutput.write() (这是输出流)
10. [干膜厚度] 改变它的状态 准备好了 悬而未决的
11. [干膜厚度] 由于上面的断点,此处暂停
12. 挂起/未读时关闭

不幸的是,在 asnyContext.complete() 检查是不够的 output.isReady() true 因为码头重新开放 HttpOutput (请参见下面的堆栈),因此您需要在侦听器中为其设置单独的标志。

Thread [jetty-13] (Suspended (access of field _state in HttpOutput))    
    HttpOutput.reopen() line: 195   
    HttpOutput.recycle() line: 923  
    Response.recycle() line: 138    
    HttpChannelOverHttp(HttpChannel).recycle() line: 269    
    HttpChannelOverHttp.recycle() line: 83  
    HttpConnection.onCompleted() line: 424  
    HttpChannelOverHttp(HttpChannel).onCompleted() line: 695    
    HttpChannelOverHttp(HttpChannel).handle() line: 493 
    HttpChannelOverHttp(HttpChannel).run() line: 293    
    QueuedThreadPool.runJob(Runnable) line: 708 
    QueuedThreadPool$2.run() line: 626  
    Thread.run() line: 748  

此外, isReady() 同时返回 真的 isReady() returns true in closed state - why?

最终的实现是类似的:

@Override
protected void doPost(final HttpServletRequest req, final HttpServletResponse resp)
        throws ServletException, IOException {
    resp.setContentType(MediaType.OCTET_STREAM.type());
    resp.setStatus(HttpServletResponse.SC_OK);
    resp.setBufferSize(4096);
    resp.flushBuffer();
    final AsyncContext async = req.startAsync();
    async.setTimeout(5_000); // millis
    final ServletOutputStream output = resp.getOutputStream();
    final QueueWriteListener writeListener = new QueueWriteListener(async, output);
    async.addListener(writeListener);
    output.setWriteListener(writeListener);
}

private static class QueueWriteListener implements AsyncListener, WriteListener {

    private static final Logger logger = LoggerFactory.getLogger(QueueWriteListener.class);

    private final AsyncContext asyncContext;
    private final ServletOutputStream output;

    @GuardedBy("this")
    private boolean completed = false;

    public QueueWriteListener(final AsyncContext asyncContext, final ServletOutputStream output) {
        this.asyncContext = checkNotNull(asyncContext, "asyncContext cannot be null");
        this.output = checkNotNull(output, "output cannot be null");
    }

    @Override
    public void onWritePossible() throws IOException {
        writeImpl();
    }

    private synchronized void writeImpl() throws IOException {
        if (completed) {
            return;
        }
        while (output.isReady()) {
            final byte[] message = getNextMessage();
            if (message == null) {
                output.flush();
                return;
            }
            output.write(message);
        }
    }

    private synchronized void completeImpl() {
        // also stops DataFeederThread to call bufferArrived
        completed = true;
        asyncContext.complete();
    }

    @Override
    public void onError(final Throwable t) {
        logger.error("Writer.onError", t);
        completeImpl();
    }

    public void dataArrived() {
        try {
            writeImpl();
        } catch (RuntimeException | IOException e) {
            ...
        }
    }

    public void noMoreData() {
        completeImpl();
    }

    @Override
    public synchronized void onComplete(final AsyncEvent event) throws IOException {
        completed = true; // might not needed but does not hurt
    }

    @Override
    public synchronized void onTimeout(final AsyncEvent event) throws IOException {
        completeImpl();
    }

    @Override
    public void onError(final AsyncEvent event) throws IOException {
        logger.error("onError", event.getThrowable());
    }

    ...
}

更新2018-08-01 :实际上它没有完全修复警告,请参见: “Closed while Pending/Unready” warnings from Jetty

您正在使用Servlet规范中的异步I/O并手动关闭响应流。

关键事实:close调用意味着写操作。

在Jetty案例中,IOException告诉您,对响应流close的手动调用导致了不希望的副作用。

所以在异步I/O模式下 ServletOutputStream.isReady() 应该是在使用前做的 ServletOutputStream.close() isReady() 是真的。

ServletOutputStream.close() AsyncContext.complete() 已经接到这封信了。

但是,如果先前的书面文件尚未完成和/或 还没打电话来,这件具体的裸体案件 ServletOutputStream.close() 允许,但行为是中止响应,放弃 ServletOutputStream .

这就是为什么你得到 IOException("Closed while Pending/Unready")

现在你应该扪心自问为什么要关闭响应流?根据Servlet规范,这不是必需的,如果您想使用 RequestDispatcher 将多个servlet响应组合在一起。