版權(quán)聲明：本文為博主原創(chuàng)文章者填，未經(jīng)博主允許不得轉(zhuǎn)載端幼。
摘要
考慮這樣一個(gè)需求，并發(fā)處理一批任務(wù)，每個(gè)任務(wù)都完成之后藻烤，對(duì)結(jié)果做一些后續(xù)處理哨颂，最后匯總結(jié)果祭钉。第一個(gè)方案：?jiǎn)?dòng)多個(gè)線(xiàn)程并發(fā)處理任務(wù)霸饲，并循環(huán)監(jiān)控每一個(gè)線(xiàn)程的處理結(jié)果Futrue，直到所有Future返回為止患亿。這個(gè)方案可行传蹈，但還需要自己監(jiān)控所有的結(jié)果完成情況押逼，是不是很乏味。來(lái)試試CompletionService吧惦界。
CompletionService
先看看這個(gè)接口定義了哪些方法：
- Future<V> submit(Callable<V> task); 提交Callable任務(wù)挑格，并返回Future結(jié)果。
- Future<V> submit(Runnable task, V result); 與上一個(gè)方法類(lèi)似沾歪，當(dāng)任務(wù)完成時(shí)返回指定的result對(duì)象漂彤。
- Future<V> take() throws InterruptedException; 獲取并移除最新完成的任務(wù)結(jié)果，該過(guò)程是阻塞的灾搏。
- Future<V> poll(); 獲取并移除最新完成的任務(wù)結(jié)果挫望，如果沒(méi)有結(jié)果則返回null。
- Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException; 指定超時(shí)時(shí)間等待獲取并移除最新的任務(wù)結(jié)果狂窑。
從獲取結(jié)果的幾個(gè)接口可以看到媳板，返回的都是最新完成的結(jié)果。這就是重點(diǎn)所在泉哈，我們可以不需要去監(jiān)控等待每一個(gè)結(jié)果（如果等待的第一個(gè)Future是最慢的蛉幸，豈不是會(huì)妨礙其他先完成的任務(wù)嗎），而是按結(jié)果完成順序得到了每一個(gè)返回結(jié)果丛晦，先完成的結(jié)果可以先繼續(xù)執(zhí)行后續(xù)處理奕纫，這不是挺好嘛。
ExecutorCompletionService是該接口的實(shí)現(xiàn)類(lèi)烫沙，內(nèi)部有一個(gè)線(xiàn)程池和BlockingQueue隊(duì)列若锁。它的實(shí)現(xiàn)原理其實(shí)挺簡(jiǎn)單：每個(gè)提交給ExecutorCompletionService的任務(wù)，都會(huì)被封裝成一個(gè)QueueingFuture（FutureTask的子類(lèi)）斧吐，它重寫(xiě)了done()方法（該方法會(huì)在任務(wù)執(zhí)行完成之后回調(diào)）煤率，將執(zhí)行完成的FutureTask加入到內(nèi)部隊(duì)列乏冀，take()等方法其實(shí)是到內(nèi)部隊(duì)列中獲取得到最新完成的結(jié)果FutrueTask辆沦。
對(duì)比
從代碼層面來(lái)看看兩種方案的差異：
public void test1() {
ExecutorService executorService = Executors.newFixedThreadPool(10);
List<Future<String>> results = new ArrayList<Future<String>>(10);
for(int i=0; i<10; i++) {
Future<String> result = executorService.submit(new MyRunnable());
results.add(result);
}
for(Future result : results) {
String str = result.get();//遍歷等待每一個(gè)Future, 如果第一個(gè)任務(wù)是最慢的肢扯，那么整個(gè)進(jìn)度就會(huì)被拖慢
//do something
}
//匯總操作
executorService.shutdown();
}
public void test2() {
ExecutorService executorService = Executors.newFixedThreadPool(10);
CompletionService<String> completionService = new executorCompletionService<String>(executorService);
for(int i=0; i<10; i++) {
completionService.submit(new MyRunnable());
}
for(int i=0; i<10; i++) {
String str = completionService.take().get();//先完成的結(jié)果先執(zhí)行后續(xù)處理
//do something
}
//匯總操作
executorService.shutdown();
}
從代碼量上看差異比較少蔚晨，但是方案2不用單獨(dú)維護(hù)一個(gè)List來(lái)保存所有的處理結(jié)果Future。重要的是多糠，completionService因?yàn)槿蝿?wù)結(jié)果按完成順序陸續(xù)到來(lái)浩考，每個(gè)任務(wù)的進(jìn)度不會(huì)相互干擾析孽，那么后續(xù)操作也不會(huì)相互影響，而第一種方案中如果第一個(gè)任務(wù)很慢闷畸，那么其他任務(wù)都要空閑等待第一個(gè)任務(wù)完成吞滞，才能繼續(xù)后面的操作，這一點(diǎn)就明顯影響到了性能裁赠。
小心踩坑
- 關(guān)閉線(xiàn)程池
在測(cè)試方法中佩捞，為了演示而創(chuàng)建了線(xiàn)程池，方法結(jié)束時(shí)也關(guān)閉了線(xiàn)程池一忱。如果你的代碼與示例代碼類(lèi)似帘营，那么請(qǐng)記住關(guān)閉線(xiàn)程池，否則即使方法退出之后问顷，創(chuàng)建的線(xiàn)程也得不到回收和關(guān)閉杜窄，遲早將耗盡資源或撐爆內(nèi)存塞耕。如果你的線(xiàn)程池是全局共享的嘴瓤，那么不存在這個(gè)問(wèn)題，JVM關(guān)閉時(shí)會(huì)關(guān)閉線(xiàn)程池胆胰。
- 錯(cuò)誤的使用方式
public void test3() {
Future<String> future = completionService.submit(new MyRunnable());//這里的線(xiàn)程池是共享的
String str = future.get();
//do something
}
該場(chǎng)景也許不太合適使用completeService蜀涨，但是這里要說(shuō)明的是另一個(gè)問(wèn)題蝎毡，直接使用completionService.submit的返回結(jié)果Future會(huì)造成內(nèi)存泄漏，因?yàn)樵摲绞街魂P(guān)心獲取當(dāng)前返回的結(jié)果沐兵，而忽略了BlockingQueue中保存的Future對(duì)象别垮，BlockingQueue隊(duì)列會(huì)不斷變大（默認(rèn)實(shí)現(xiàn)是LinkedBlockingQueue扎谎，無(wú)界隊(duì)列）碳想，遲早將內(nèi)存撐爆毁靶。正確的使用方式還是通過(guò)completionService.take()來(lái)獲取Future對(duì)象胧奔。
如有什么地方描述不對(duì)预吆，歡迎指出龙填。
