使用 EntityUtils.consume(httpEntity) 消耗到流结束如何导致将连接释放回连接池

Question

我在这里读到EntityUtils.consume(httpEntity)将导致将连接释放回连接池，但是当我查看源代码时，我不明白这是怎么回事。有人可以指出我在使用低级弹性搜索 Rest 客户端时EntityUtils.consume(httpEntity)或EntityUtils.toString(httpEntity)释放连接的代码部分吗？

如果有SocketTimeoutException并且我不使用HttpEntity，连接会发生什么情况？

Answer 1

客户端关闭和连接到池的连接释放(步骤)

1. EntityUtils.consume&EntityUtils.toString>，如果第一个完全消耗实体，它将close()instream。第二个将始终在其finally子句中调用instream.close()。instream是输入流变量的名称。

2. instream.close()> 对于此示例，InputStream的实现是一个ContentInputStream。close()方法强制通过代码段中显示的循环机制读取ContentInputStream直到其结束。

   在此流调用之后将导致EOF异常。

   @Override
   public void close() throws IOException 
   {
   final byte tmp[] = new byte[1024];
   /*loop until read() is -1, which means, advance the buffer till is end*/
   while (this.buffer.read(tmp, 0, tmp.length) >= 0) {}
   super.close();
   }
   

3. Pool> 检查所有池资源状态。此操作可能由某些操作(作为新请求)触发，也可能由基础线程管理。如果一个资源/流被另一端关闭，它将得到一个EOF异常(因为缓冲区被迫前进到最后)。该点被标记为无效。

4. Pool> 所有无效斑点都会被回收。它将删除关闭的流并创建新的流，或恢复现有流而无需擦除+创建(取决于资源类型)。这意味着保存流的位置再次可用，并准备好使用新流：

   连接将释放回池。另一端不再使用它，因此池可以完全控制它。现在允许池擦除它，恢复它，并将其分配给另一个请求者。

<小时 />

示例

让我们想象一个管理 3 个资源的Pool，例如HttpConnections. 您已经有 3 个线程使用此池，因此所有位置都被占用了。

同时，ThreadZ等待连接释放回池

(spot1) [HttpConn1] -- ThreadA
(spot2) [HttpConn2] -- ThreadB
(spot3) [HttpConn3] -- ThreadC

ThreadA完成了工作并关闭了连接。当PoolEntry的状态为关闭时，Pool会注意到这一点。不同的PoolEntry实现会检查这是不同的方式，其中之一是在尝试从流中读取时获得EOF异常。其他实现可能具有不同的机制来检查资源是否已关闭。如果PoolEntry告诉他的资源已关闭/无效，Pool将回收此位置。这里有两个选项：

a) 擦除并创建。

(spot1) [HttpConn4] // the pool removes the old one and creates a new connection
(spot2) [HttpConn2] -- ThreadB
(spot3) [HttpConn3] -- ThreadC

b) 恢复。

(spot1) [HttpConn1] // the pool is able to "reset" the existing resource
(spot2) [HttpConn2] -- ThreadB
(spot3) [HttpConn3] -- ThreadC

">

重新释放连接"可以翻译为"现在又有可用的点/资源了"。池现在可以连接到ThreadZ：

(spot1) [HttpConn1] -- ThreadZ
(spot2) [HttpConn2] -- ThreadB
(spot3) [HttpConn3] -- ThreadC

<小时 />

consume/toString- 连接释放

上述所有说明都意味着在InputStream中调用close()将触发连接释放。

这在consume(如果实体内容已完全使用)和toString方法中都会发生：

public static void consume(final HttpEntity entity) throws IOException 
{ 
if (entity == null) 
return; 

if (entity.isStreaming()) 
{ 
InputStream instream = entity.getContent(); 
if (instream != null) 
instream.close();   // <-- connection release
} 
} 
public static String toString(final HttpEntity entity, final Charset defaultCharset) 
throws IOException, ParseException 
{ 
Args.notNull(entity, "Entity"); 
InputStream instream = entity.getContent(); 
if (instream == null) { 
return null; 
} 
try { 
Args.check(entity.getContentLength() <= Integer.MAX_VALUE,  
"HTTP entity too large to be buffered in memory"); 
int i = (int)entity.getContentLength(); 
if (i < 0) { 
i = 4096; 
} 
Charset charset = null; 
try { 
ContentType contentType = ContentType.getOrDefault(entity); 
charset = contentType.getCharset(); 
} catch (UnsupportedCharsetException ex) { 
throw new UnsupportedEncodingException(ex.getMessage()); 
} 
if (charset == null) { 
charset = defaultCharset; 
} 
if (charset == null) { 
charset = HTTP.DEF_CONTENT_CHARSET; 
} 
Reader reader = new InputStreamReader(instream, charset); 
CharArrayBuffer buffer = new CharArrayBuffer(i); 
char[] tmp = new char[1024]; 
int l; 
while((l = reader.read(tmp)) != -1) { 
buffer.append(tmp, 0, l); 
} 
return buffer.toString(); 
} finally { 
instream.close();     // <--- connection release
} 
} 

---

如果存在 SocketTimeoutException 并且我不使用 HttpEntity，连接会发生什么情况？

正如你所注意到的，这两种方法都抛出了一个IOException，并且SocketTimeoutException是从它继承的。调用方有责任捕获此异常，并在发生这种情况时设法关闭所有资源。例如：

void tryConsume()
{
try 
{
//...
EntityUtils.consume(httpEntity);
//...
}
catch (IOException)
{
//SocketTimeoutException happened. Log the error,etc
// (Close resources here...)
}
finally
{
//...Or maybe include a finally clause and close them here, if you wish 
// them to be closed regardless of success/failure.
if (httpEntity!=null)
{
InputStream instream = httpEntity.getContent(); 
if (instream != null) 
instream.close();   /* <-- connection release. when checking this 
spot, the pool will get (f.e) an EOF 
exception. This will lead to replacing this 
resource with a fresh new connection and 
setting the spot status as avaliable. */
}
}
}

请注意，如果抛出SocketTimeoutException，特定的PoolEntry实现还可以检查资源是否无效，而无需close()调用。使用close()可以保证Pool在正确使用后回收该点，并且当您能够捕获抛出的异常时，可以用作"无效标记"。

但是特定的Pool实现也将能够检查资源是否无效，即使未捕获的Exception不允许您专门调用close()，因为它们将能够使用不同的机制检查状态。例如，检查连接处于IDLE状态的次数。如果这个时间优于Pool标记的某个侵入，则该位置将被回收，而无需客户事先close()电话。

这一次，Pool将是调用close()的结束，避免客户端可能出现deadlock，如果这个客户端不管理最大连接时间或某些异常。