您的位置: 首页  >  文章  >  Tomcat源码分析之：ServletOutputStream的实现

Tomcat源码分析之：ServletOutputStream的实现

分类: 文章 • 2024-12-14 15:58:40

转自：https://blog.****.net/fjslovejhl/article/details/23939017打开

貌似很久都没有写博客了，tomcat8的代码已经看了很多，主体部分的代码也都看得差不多了，发现在tomcat8中已经完全支持非阻塞的方式接收以及发送数据了。。。。但是比较遗憾的是，以前遗留下来的太多的老代码都不支持这种新的方式来发送数据。。。木有办法。。。

这里来看看Tomcat中是如何实现ServletOutputStream的吧。。。。

在具体的来看它之前，这里先来一张图来描述一下Tomcat的数据发送时候的流动。。。

Tomcat源码分析之：ServletOutputStream的实现


这张图形已经比较好的展现成了Tomcat中对IO这一块封装的层次关系吧，首先是最上层的

ServletOutputStream对象，它是用户代码可以接触到的对象，它自带了自己的OutputBuffer，这里写数据其实是通过这个这个buffer完成的，数据是写到这个OutputBuffer里面

接下来的一层就是Tomcat内部的Response类型了，每一个ServletResponse对象都对应着一个唯一的Tomcat内部的response类型，当然了，他也有自己的buffer，InternalNioOutputBuffer类型的对象，这里上层ServletOutputStream写数据将会流动到这里。。。

最下层就是与channel关联的部分了，它也有自己的buffer，这里一般就是java类库中的niobuffer，上层的数据将会流动到这里。。。

最后才是将数据通过channel发送出去。。。

嗯，虽然好像层次稍微多了些，而且刚刚开始看代码就觉得稍微有点繁琐。。不过当将这个层次关系理清楚了之后还是蛮简单的。。。

好啦，接下来来看代码了。。。

这里就通过一次write来跟踪整个代码的执行轨迹吧，首先来看看CoyoteOutputStream类型的write方法：

[java] view plain copy
  1. //发送一个byte数组的一部分数据  
  2. public void write(byte[] b, int off, int len) throws IOException {  
  3.     boolean nonBlocking = checkNonBlockingWrite();  //判断是否支持非阻塞的发送数据，这里判断的标准其实是用户代码是否加入了WriteListener  
  4.     ob.write(b, off, len);  //将数据写到outputbuffer里面  
  5.     if (nonBlocking) {  
  6.         checkRegisterForWrite();  //如果是非阻塞的发送数据的话，需要确保channel有注册写事件  
  7.     }  
  8. }  

这里首先要判断是否支持非阻塞的发送数据，这里就不细讲这部分的类容了，，，来看看这里的outputbuffer的write行为吧：

[java] view plain copy
  1. //写入一个字节数组的一部分  
  2.     public void write(byte b[], int off, int len) throws IOException {  
  3.   
  4.         if (suspended) {  
  5.             return;  
  6.         }  
  7.   
  8.         writeBytes(b, off, len);  
  9.   
  10.     }  
  11.   
  12.   
  13.     private void writeBytes(byte b[], int off, int len)  
  14.         throws IOException {  
  15.   
  16.         if (closed) {  //如果已经关闭了，那么直接返回吧  
  17.             return;  
  18.         }  
  19.         //这里其实是写到buffer里，如果数据过大，也有可能调用realWriteBytes方法真正的调用底层写数据  
  20.         bb.append(b, off, len);  
  21.         bytesWritten += len;  
  22.   
  23.         // if called from within flush(), then immediately flush  
  24.         // remaining bytes  
  25.         if (doFlush) {  
  26.             bb.flushBuffer();  
  27.         }  
  28.   
  29.     }  

这里其实就是将数据直接放到当年的bytechunk就好了。。。但是这里如果数据比较多的话，会将数据直接写到下一层，也就是tomcat内置的response。。。来看看bytechunk的append方法吧：

[java] view plain copy
  1. //加入一个字节数组的一部分数据  
  2. public void append( byte src[], int off, int len )  
  3.     throws IOException  
  4. {  
  5.     // will grow, up to limit  
  6.     makeSpace( len );  //确保有这么多空间可以写  
  7.   
  8.     // if we don't have limit: makeSpace can grow as it wants  
  9.     if( limit < 0 ) {  //表示没有空间限制  
  10.         // assert: makeSpace made enough space  
  11.         System.arraycopy( src, off, buff, end, len );  //将数据复制过来  
  12.         end+=len;  //将end偏移加上len  
  13.         return;  
  14.     }  
  15.   
  16.     // Optimize on a common case.  
  17.     // If the buffer is empty and the source is going to fill up all the  
  18.     // space in buffer, may as well write it directly to the output,  
  19.     // and avoid an extra copy  
  20.     //如果一次就填满了，那么还是写出去好了  
  21.     if ( len == limit && end == start && out != null ) {  
  22.         out.realWriteBytes( src, off, len );  //因为要写的数据比较大，所以直接写到更下层去  
  23.         return;  
  24.     }  
  25.     // if we have limit and we're below  
  26.     if( len <= limit - end ) {  //表示还有足够的空间可以写数据  
  27.         // makeSpace will grow the buffer to the limit,  
  28.         // so we have space  
  29.         System.arraycopy( src, off, buff, end, len );  
  30.         end+=len;  
  31.         return;  
  32.     }  
  33.   
  34.     // need more space than we can afford, need to flush  
  35.     // buffer  
  36.   
  37.     // the buffer is already at ( or bigger than ) limit  
  38.   
  39.     // We chunk the data into slices fitting in the buffer limit, although  
  40.     // if the data is written directly if it doesn't fit  
  41.   
  42.     //代码执行到这里，说明空间不够了  
  43.     int avail=limit-end;  //还剩下多大的空间可以写数据  
  44.     System.arraycopy(src, off, buff, end, avail);  
  45.     end += avail;  
  46.   
  47.     flushBuffer();   
  48.   
  49.     int remain = len - avail;  
  50.   
  51.     while (remain > (limit - end)) {  //不断的尝试写数据到下面去  
  52.         out.realWriteBytes( src, (off + len) - remain, limit - end );  
  53.         remain = remain - (limit - end);  
  54.     }  
  55.   
  56.     System.arraycopy(src, (off + len) - remain, buff, end, remain);  
  57.     end += remain;  
  58.   
  59. }  

这里可以看到realWriteBytes方法，它其实就是外面的outputbuffer定义的方法，来看看吧：

[java] view plain copy
  1. //这个才是真正的调用底层发送数据，其实又是调用tomcat的response来写数据  
  2. public void realWriteBytes(byte buf[], int off, int cnt)  
  3.         throws IOException {  
  4.   
  5.     if (closed) {  
  6.         return;  
  7.     }  
  8.     if (coyoteResponse == null) {  
  9.         return;  
  10.     }  
  11.   
  12.     // If we really have something to write  
  13.     if (cnt > 0) {  
  14.         // real write to the adapter  
  15.         outputChunk.setBytes(buf, off, cnt);  //设置outputchunk  
  16.         try {  
  17.             coyoteResponse.doWrite(outputChunk);  //通过tomcat的response来写数据，其实是写到httpprocessor的buffer里面去了  
  18.         } catch (IOException e) {  
  19.             // An IOException on a write is almost always due to  
  20.             // the remote client aborting the request.  Wrap this  
  21.             // so that it can be handled better by the error dispatcher.  
  22.             throw new ClientAbortException(e);  
  23.         }  
  24.     }  
  25.   
  26. }  

嗯，这里就与上面的层次对应上了吧，其实就是写到tomcat内置的response。。好了，接下来继续。。

[java] view plain copy
  1. //在servlet的outputStream可能会调用这个方法来写数据  
  2. public void doWrite(ByteChunk chunk/*byte buffer[], int pos, int count*/)  
  3.     throws IOException  
  4. {  
  5.     outputBuffer.doWrite(chunk, this);  //调用在httpprocessor里面创建的outputbuffer来写数据，这里会将数据写到niochannel的buffer，然后最终发送出去  
  6.     contentWritten+=chunk.getLength();   //标记已经发送的数据量的大小  
  7. }  

嗯，这里其实是写到internalNiobuffer里去。。。。继续看吧：

[java] view plain copy
  1. public int doWrite(ByteChunk chunk, Response res) throws IOException {  
  2.   
  3.            int len = chunk.getLength();  
  4.            int start = chunk.getStart();  
  5.            byte[] b = chunk.getBuffer();  
  6.            addToBB(b, start, len);  
  7.            byteCount += chunk.getLength();    
  8.            return chunk.getLength();  
  9.        }  

嗯，没啥意思，继续：

[java] view plain copy
  1. private synchronized void addToBB(byte[] buf, int offset, int length)  
  2.         throws IOException {  
  3.   
  4.     if (length == 0) return;  
  5.   
  6.     // Try to flush any data in the socket's write buffer first  
  7.     //首先尝试先将数据发送出去  
  8.     boolean dataLeft = flushBuffer(isBlocking());  
  9.   
  10.     // Keep writing until all the data is written or a non-blocking write  
  11.     // leaves data in the buffer  
  12.     //这里只有在缓冲区里面已经没有数据了才继续发送  
  13.     while (!dataLeft && length > 0) {  
  14.         //首先将要发送的数据copy到niochanel的发送buffer里面去  
  15.         int thisTime = transfer(buf,offset,length,socket.getBufHandler().getWriteBuffer());  
  16.         length = length - thisTime;  //计算还剩下多少字节没有写到niochannel的buffer里面，其实这里也就当做将数据转移到了niochannel的buffer就算是写出去了  
  17.         offset = offset + thisTime;  //这里用于调整偏移量  
  18.         //这里调用writeToSocket方法将niochannel的buffer的里面的数据通过socket写出去  
  19.         int written = writeToSocket(socket.getBufHandler().getWriteBuffer(),  
  20.                 isBlocking(), true);  //如果在tomcat的response里面有writelistener的话，可以异步的写  
  21.         if (written == 0) {  //都没有写出去字节  
  22.             dataLeft = true;  
  23.         } else {  
  24.             dataLeft = flushBuffer(isBlocking());  //flush一下，看一下是否还会有数据剩余  
  25.         }  
  26.     }  
  27.   
  28.     NioEndpoint.KeyAttachment ka = (NioEndpoint.KeyAttachment)socket.getAttachment(false);  
  29.     if (ka != null) ka.access();//prevent timeouts for just doing client writes  
  30.   
  31.     if (!isBlocking() && length > 0) { //在非阻塞的发送中，如果实在发送不出去，需要保存在额外的buffer里面  
  32.         // Remaining data must be buffered  
  33.         addToBuffers(buf, offset, length);  
  34.     }  
  35. }  

这里其实主要是调用flushBuffer方法，将数据传给下层的niochannel，而且可以看到对于过多的数据这里还会 做一层缓存。。。。

[java] view plain copy
  1. //这里其实是flush  niochannel的buffer  
  2. protected boolean flushBuffer(boolean block) throws IOException {  
  3.   
  4.     //prevent timeout for async,  
  5.     SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());  
  6.     if (key != null) {  
  7.         NioEndpoint.KeyAttachment attach = (NioEndpoint.KeyAttachment) key.attachment();  
  8.         attach.access();  
  9.     }  
  10.   
  11.     boolean dataLeft = hasMoreDataToFlush();   //其实这里是判断niochannel的buffer里面是否还有数据要写  
  12.   
  13.     //write to the socket, if there is anything to write  
  14.     if (dataLeft) {  //如果niochannel的buffer里面还有数据发送，那么继续写  
  15.         writeToSocket(socket.getBufHandler().getWriteBuffer(),block, !flipped);  
  16.     }  
  17.   
  18.     dataLeft = hasMoreDataToFlush();  
  19.   
  20.     //这里如果niochannel的buffer里面的数据已经发送完了，那么将将以前缓冲的数据再发送出去  
  21.     if (!dataLeft && bufferedWrites.size() > 0) {  
  22.         Iterator<ByteBufferHolder> bufIter = bufferedWrites.iterator();  //遍历待发送的数据  
  23.         while (!hasMoreDataToFlush() && bufIter.hasNext()) {  
  24.             ByteBufferHolder buffer = bufIter.next();  
  25.             buffer.flip();  
  26.             while (!hasMoreDataToFlush() && buffer.getBuf().remaining()>0) {  
  27.                 transfer(buffer.getBuf(), socket.getBufHandler().getWriteBuffer());  
  28.                 if (buffer.getBuf().remaining() == 0) {  //如果当前buffer里面的所有数据都已经转移到了niochannel的buffer里面，那么可以将这个buffer移除了  
  29.                     bufIter.remove();  
  30.                 }  
  31.                 writeToSocket(socket.getBufHandler().getWriteBuffer(),block, true);  
  32.                 //here we must break if we didn't finish the write  
  33.             }  
  34.         }  
  35.     }  
  36.   
  37.     return hasMoreDataToFlush();  
  38. }  

嗯，这里其实主要是调用writeToSocket方法。。。来看看吧：

[java] view plain copy
  1. //这里其实调用socket来写数据  
  2. private synchronized int writeToSocket(ByteBuffer bytebuffer, boolean block, boolean flip) throws IOException {  
  3.     if ( flip ) {  
  4.         bytebuffer.flip();  
  5.         flipped = true;  
  6.     }  
  7.   
  8.     int written = 0;  
  9.     NioEndpoint.KeyAttachment att = (NioEndpoint.KeyAttachment)socket.getAttachment(false);  
  10.     if ( att == null ) throw new IOException("Key must be cancelled");  
  11.     long writeTimeout = att.getWriteTimeout();  
  12.     Selector selector = null;  
  13.     try {  
  14.         selector = pool.get();  
  15.     } catch ( IOException x ) {  
  16.         //ignore  
  17.     }  
  18.     try {  
  19.         written = pool.write(bytebuffer, socket, selector, writeTimeout, block);  
  20.         //make sure we are flushed  
  21.         do {  
  22.             //对于niochanel，这个flush方法其实是没用的  
  23.             if (socket.flush(true,selector,writeTimeout)) break;  
  24.         }while ( true );  
  25.     }finally {  
  26.         if ( selector != null ) {  
  27.             pool.put(selector);  
  28.         }  
  29.     }  
  30.     if ( block || bytebuffer.remaining()==0) {  
  31.         //blocking writes must empty the buffer  
  32.         //and if remaining==0 then we did empty it  
  33.         bytebuffer.clear();  
  34.         flipped = false;  
  35.     }  
  36.     // If there is data left in the buffer the socket will be registered for  
  37.     // write further up the stack. This is to ensure the socket is only  
  38.     // registered for write once as both container and user code can trigger  
  39.     // write registration.  
  40.     return written;  
  41. }  

这里因为涉及到了一些阻塞的或者非阻塞的发送数据。。所以可能会用到selector。。。

[java] view plain copy
  1. public int write(ByteBuffer buf, NioChannel socket, Selector selector,  
  2.                  long writeTimeout, boolean block) throws IOException {  
  3.     if ( SHARED && block ) { //对于写数据，一般都是这里  
  4.         return blockingSelector.write(buf,socket,writeTimeout);  
  5.     }  
  6.     //但是如果有outputstream的listener的话，可以采用非阻塞的方式来发送大量的数据  
  7.     SelectionKey key = null;  
  8.     int written = 0;  
  9.     boolean timedout = false;  
  10.     int keycount = 1; //assume we can write  //假装刚开始是可以写的  
  11.     long time = System.currentTimeMillis(); //start the timeout timer  
  12.     try {  
  13.         while ( (!timedout) && buf.hasRemaining() ) {  
  14.             int cnt = 0;  
  15.             if ( keycount > 0 ) { //only write if we were registered for a write  
  16.                 cnt = socket.write(buf); //write the data  
  17.                 if (cnt == -1) throw new EOFException();  //出错了  
  18.   
  19.                 written += cnt;  
  20.                 if (cnt > 0) {  
  21.                     time = System.currentTimeMillis(); //reset our timeout timer  
  22.                     continue; //we successfully wrote, try again without a selector  
  23.                 }  
  24.                 if (cnt==0 && (!block)) {   //这里对于非阻塞的写，就直接返回了  
  25.                     break; //don't block  
  26.                 }  
  27.             }  
  28.             if ( selector != null ) {  
  29.                 //register OP_WRITE to the selector  
  30.                 if (key==null) key = socket.getIOChannel().register(selector, SelectionKey.OP_WRITE);  
  31.                 else key.interestOps(SelectionKey.OP_WRITE);  
  32.                 if (writeTimeout==0) {  
  33.                     timedout = buf.hasRemaining();  
  34.                 } else if (writeTimeout<0) {  
  35.                     keycount = selector.select();  
  36.                 } else {  
  37.                     keycount = selector.select(writeTimeout);  
  38.                 }  
  39.             }  
  40.             if (writeTimeout > 0 && (selector == null || keycount == 0) ) timedout = (System.currentTimeMillis()-time)>=writeTimeout;  
  41.         }//while  
  42.         if ( timedout ) throw new SocketTimeoutException();  
  43.     } finally {  
  44.         if (key != null) {  
  45.             key.cancel();  
  46.             if (selector != null) selector.selectNow();//removes the key from this selector  
  47.         }  
  48.     }  
  49.     return written;  
  50. }  

这里如果就直接调用niochannel来发送数据了。。不过其实这里还会涉及到将数据转移到niochannel的buffer。。然后才发送数据。。。


到这里整个数据的流动层次对照着上面的图形应该就算是 比较明白了吧。。。。


相关推荐