Jetty 8長(zhǎng)連接上的又一個(gè)坑
2018-07-20 來源：編程學(xué)習(xí)網(wǎng)
Jetty 8 長(zhǎng)連接的超時(shí)斷開連接的機(jī)制：超時(shí)連接機(jī)制針對(duì)IO傳輸過程中的數(shù)據(jù)阻塞時(shí)間超過一定閾值時(shí)，斷開該連接。阻塞指當(dāng)前處于數(shù)據(jù)傳輸階段，但是連續(xù)指定時(shí)間內(nèi)都沒有發(fā)出或者接收到任何數(shù)據(jù)時(shí)，Jetty系統(tǒng)斷開該連接。強(qiáng)調(diào)一下，只有在數(shù)據(jù)傳輸過程中才會(huì)有超時(shí)機(jī)制。在服務(wù)端處理已經(jīng)收到的數(shù)據(jù)時(shí)是不會(huì)檢測(cè)該超時(shí)時(shí)間的。
下面看一下具體的代碼實(shí)現(xiàn)。在jetty 8.1.17版本中，由以下代碼控制一個(gè)連接的空閑、非空閑和斷開檢查方法，在SelectChannelEndpoint類中：
/* ------------------------------------------------------------ */ public void setCheckForIdle(boolean check) { if (check) { _idleTimestamp=System.currentTimeMillis(); _checkIdle=true; } else _checkIdle=false; } /* ------------------------------------------------------------ */ public boolean isCheckForIdle() { return _checkIdle; } /* ------------------------------------------------------------ */ protected void notIdle() { _idleTimestamp=System.currentTimeMillis(); } /* ------------------------------------------------------------ */ public void checkIdleTimestamp(long now) { if (isCheckForIdle() && _maxIdleTime>0) { final long idleForMs=now-_idleTimestamp; if (idleForMs>_maxIdleTime) { // Don't idle out again until onIdleExpired task completes. setCheckForIdle(false); _manager.dispatch(new Runnable() { public void run() { try { onIdleExpired(idleForMs); } finally { setCheckForIdle(true); } } }); } } }
幾個(gè)關(guān)鍵點(diǎn)地方：當(dāng)數(shù)據(jù)傳輸?shù)倪^程中，發(fā)現(xiàn)無法接收到和寫出數(shù)據(jù)時(shí)，會(huì)調(diào)用setCheckForIdle(true)方法，從當(dāng)前時(shí)間點(diǎn)開始計(jì)時(shí)，當(dāng)后臺(tái)select線程發(fā)現(xiàn)該連接的空閑時(shí)間達(dá)到閾值時(shí)，則調(diào)用onIdleExpired方法。還有一種場(chǎng)景是，在一個(gè)請(qǐng)求結(jié)束后，立即將該請(qǐng)求置為空閑狀態(tài)。直到連接關(guān)閉或者該連接上面來了新的請(qǐng)求。另外，每個(gè)新的連接建立時(shí)，會(huì)在構(gòu)造函數(shù)中默認(rèn)調(diào)用一次該方法設(shè)置連接為空閑狀態(tài)。
在哪些情況下會(huì)調(diào)用相反的設(shè)置呢，即將該連接置為非空閑狀態(tài)的setCheckForIdle(false)方法，和刷新當(dāng)前的idle時(shí)間方法notIdle()呢？第一個(gè)方法每次收到一個(gè)請(qǐng)求的數(shù)據(jù)提交后端的servlet的時(shí)候調(diào)用，后一個(gè)方法在每次刷出或者讀到數(shù)據(jù)時(shí)調(diào)用。這樣確保后端的servlet在處理數(shù)據(jù)時(shí)，不至于因?yàn)樘幚頃r(shí)間過長(zhǎng)而被自己的select線程給關(guān)閉了。
這一次jetty的bug正是出在上述的每個(gè)請(qǐng)求的數(shù)據(jù)收集完成進(jìn)入后端處理之前發(fā)生的。看如下代碼：
AsyncHttpConnection類中，handle方法：
@Override public Connection handle() throws IOException { Connection connection = this; boolean some_progress=false; boolean progress=true; try { setCurrentConnection(this); // don't check for idle while dispatched (unless blocking IO is done). _asyncEndp.setCheckForIdle(false); // While progress and the connection has not changed while (progress && connection==this) { progress=false; try { // Handle resumed request if (_request._async.isAsync()) { if (_request._async.isDispatchable()) handleRequest(); } // else Parse more input else if (!_parser.isComplete() && _parser.parseAvailable()) progress=true; // Generate more output if (_generator.isCommitted() && !_generator.isComplete() && !_endp.isOutputShutdown() && !_request.getAsyncContinuation().isAsyncStarted()) if (_generator.flushBuffer()>0) progress=true; // Flush output _endp.flush(); // Has any IO been done by the endpoint itself since last loop if (_asyncEndp.hasProgressed()) progress=true; } catch (HttpException e) { if (LOG.isDebugEnabled()) { LOG.debug("uri="+_uri); LOG.debug("fields="+_requestFields); LOG.debug(e); } progress=true; _generator.sendError(e.getStatus(), e.getReason(), null, true); } finally { some_progress|=progress; // Is this request/response round complete and are fully flushed? boolean parserComplete = _parser.isComplete(); boolean generatorComplete = _generator.isComplete(); boolean complete = parserComplete && generatorComplete; if (parserComplete) { if (generatorComplete) { // Reset the parser/generator progress=true; // look for a switched connection instance? if (_response.getStatus()==HttpStatus.SWITCHING_PROTOCOLS_101) { Connection switched=(Connection)_request.getAttribute("org.eclipse.jetty.io.Connection"); if (switched!=null) connection=switched; } reset(); // TODO Is this still required? if (!_generator.isPersistent() && !_endp.isOutputShutdown()) { LOG.warn("Safety net oshut!!! IF YOU SEE THIS, PLEASE RAISE BUGZILLA"); _endp.shutdownOutput(); } } else { // We have finished parsing, but not generating so // we must not be interested in reading until we // have finished generating and we reset the generator _readInterested = false; LOG.debug("Disabled read interest while writing response {}", _endp); } } if (!complete && _request.getAsyncContinuation().isAsyncStarted()) { // The request is suspended, so even though progress has been made, // exit the while loop by setting progress to false LOG.debug("suspended {}",this); progress=false; } } } } finally { setCurrentConnection(null); // If we are not suspended if (!_request.getAsyncContinuation().isAsyncStarted()) { // return buffers _parser.returnBuffers(); _generator.returnBuffers(); // reenable idle checking unless request is suspended _asyncEndp.setCheckForIdle(true); } // Safety net to catch spinning if (some_progress) _total_no_progress=0; else { _total_no_progress++; if (NO_PROGRESS_INFO>0 && _total_no_progress%NO_PROGRESS_INFO==0 && (NO_PROGRESS_CLOSE<=0 || _total_no_progress< NO_PROGRESS_CLOSE)) LOG.info("EndPoint making no progress: "+_total_no_progress+" "+_endp+" "+this); if (NO_PROGRESS_CLOSE>0 && _total_no_progress==NO_PROGRESS_CLOSE) { LOG.warn("Closing EndPoint making no progress: "+_total_no_progress+" "+_endp+" "+this); if (_endp instanceof SelectChannelEndPoint) ((SelectChannelEndPoint)_endp).getChannel().close(); } } } return connection; }
可以看到，在handle方法進(jìn)入時(shí)，調(diào)用了一次：
// don't check for idle while dispatched (unless blocking IO is done). _asyncEndp.setCheckForIdle(false);
如果當(dāng)前連接是一個(gè)短連接，那么這里調(diào)用完全沒問題。請(qǐng)求處理完成后本來就可能立即斷開連接。但是如果是一個(gè)長(zhǎng)連接，該連接在處理完請(qǐng)求后，可能“休息”一段時(shí)間繼續(xù)處理新的請(qǐng)求，那么就問題就來了，從該代碼看，jetty在handle方法的while循環(huán)中處理多個(gè)請(qǐng)求，這樣可以避免同一個(gè)連接上面的多個(gè)請(qǐng)求被分到不同的線程中處理，而是綁定在一個(gè)線程上面處理，當(dāng)長(zhǎng)連接上面的請(qǐng)求比較“密集”（請(qǐng)求之間間隔極短）時(shí)，該while會(huì)循環(huán)多次，有兩種情況會(huì)進(jìn)入該請(qǐng)求：1、一個(gè)請(qǐng)求上面的數(shù)據(jù)沒有處理完，即
// else Parse more input else if (!_parser.isComplete() && _parser.parseAvailable()) progress=true;
這個(gè)代碼控制的。
另外當(dāng)一個(gè)請(qǐng)求處理完了，也會(huì)在finally里面走到progess=true上面。
// Is this request/response round complete and are fully flushed? boolean parserComplete = _parser.isComplete(); boolean generatorComplete = _generator.isComplete(); boolean complete = parserComplete && generatorComplete; if (parserComplete) { if (generatorComplete) { // Reset the parser/generator progress=true;
由這個(gè)控制。
問題出在第二個(gè)上面，當(dāng)一個(gè)請(qǐng)求處理完成后，連接會(huì)被置為空閑狀態(tài)。但是這里將progess設(shè)置為true，那么while循環(huán)立即準(zhǔn)備讀取下一個(gè)請(qǐng)求的數(shù)據(jù)，但是并沒有將連接置為非空閑狀態(tài)，此時(shí)如果服務(wù)端進(jìn)入耗時(shí)較長(zhǎng)的處理流程，那么可能不等到客戶端超時(shí)，連接就被后臺(tái)檢查空閑連接的線程斷開了。
因此這里很明顯，jetty有bug，應(yīng)該在最后的這段代碼出補(bǔ)充
// don't check for idle while dispatched (unless blocking IO is done). _asyncEndp.setCheckForIdle(false);
這個(gè)調(diào)用�；蛘呤窃诿看芜M(jìn)入while循環(huán)時(shí)調(diào)用，而不是只在進(jìn)入handle時(shí)調(diào)用。
該問題發(fā)生有幾個(gè)關(guān)鍵點(diǎn)：長(zhǎng)連接上面持續(xù)不斷有新請(qǐng)求過來，并且新請(qǐng)求發(fā)起的時(shí)間距離上一個(gè)請(qǐng)求完成的時(shí)間間隔非常短。經(jīng)過實(shí)測(cè)，python的http客戶端在處理長(zhǎng)連接上面，請(qǐng)求間隔非常短。而其他語(yǔ)言和庫(kù)編寫的客戶端測(cè)試程序都有比較長(zhǎng)的間隔，導(dǎo)致問題不易重現(xiàn)。附一個(gè)jetty的簡(jiǎn)易http長(zhǎng)連接測(cè)試程序：
import httplib count=0 conn = httplib.HTTPConnection("127.0.0.1", timeout=600) while (count < 1000000): conn.request("PUT","/") res = conn.getresponse() print res.status, res.reason print res.read() count += 1
在jetty上面講超時(shí)時(shí)間配置盡可能短，在servlet里面處理請(qǐng)求時(shí)休眠一個(gè)大于等于超時(shí)時(shí)間的值，配合上述客戶端，很容易重現(xiàn)問題。
文章來源：http://codefine.co/2822.html
版權(quán)申明：本站文章部分自網(wǎng)絡(luò)，如有侵權(quán)，請(qǐng)聯(lián)系：west999com@outlook.com
特別注意：本站所有轉(zhuǎn)載文章言論不代表本站觀點(diǎn)！
本站所提供的圖片等素材，版權(quán)歸原作者所有，如需使用，請(qǐng)與原作者聯(lián)系。
