摘要:抽象類(lèi),實(shí)現(xiàn)了的接口。將任務(wù)封裝成提交任務(wù)主要方法在任務(wù)是否超時(shí)超時(shí)時(shí)間任務(wù)書(shū)用于存放結(jié)果的,先完成的放前面。
AbstractExecutorService抽象類(lèi),實(shí)現(xiàn)了ExecutorService的接口。
newTaskFor將任務(wù)封裝成FutureTask
protectedsubmitRunnableFuture newTaskFor(Runnable runnable, T value) { return new FutureTask (runnable, value); } protected RunnableFuture newTaskFor(Callable callable) { return new FutureTask (callable); }
提交任務(wù)
public Future> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFutureinvokeAnyftask = newTaskFor(task, null); execute(ftask); return ftask; } public Future submit(Runnable task, T result) { if (task == null) throw new NullPointerException(); RunnableFuture ftask = newTaskFor(task, result); execute(ftask); return ftask; } public Future submit(Callable task) { if (task == null) throw new NullPointerException(); RunnableFuture ftask = newTaskFor(task); execute(ftask); return ftask; }
主要方法在doInvokeAny
//tasks任務(wù) //timed是否超時(shí) //nanos超時(shí)時(shí)間 privateinvokeAllT doInvokeAny(Collection extends Callable > tasks, boolean timed, long nanos) throws InterruptedException, ExecutionException, TimeoutException { if (tasks == null) throw new NullPointerException(); int ntasks = tasks.size();//任務(wù)書(shū) if (ntasks == 0) throw new IllegalArgumentException(); ArrayList > futures = new ArrayList >(ntasks); //用于存放結(jié)果的,先完成的放前面。所以第一個(gè)任務(wù)沒(méi)完成的時(shí)候,會(huì)繼續(xù)提交后續(xù)任務(wù) ExecutorCompletionService ecs = new ExecutorCompletionService (this); try { //異常信息 ExecutionException ee = null; //過(guò)期時(shí)間 final long deadline = timed ? System.nanoTime() + nanos : 0L; Iterator extends Callable > it = tasks.iterator();//獲取第一個(gè)任務(wù) 提交任務(wù) futures.add(ecs.submit(it.next())); --ntasks;//因?yàn)樘峤涣艘粋€(gè),任務(wù)數(shù)-1 int active = 1;//正在執(zhí)行的任務(wù) for (;;) { Future f = ecs.poll(); if (f == null) {//第一個(gè)沒(méi)完成 if (ntasks > 0) {//還有沒(méi)提交的任務(wù) --ntasks;//任務(wù)數(shù)-1 futures.add(ecs.submit(it.next()));//提交任務(wù) ++active;//正在執(zhí)行的任務(wù)+1 } else if (active == 0)//當(dāng)前沒(méi)任務(wù)了,但是都失敗了,異常被捕獲了 break; else if (timed) { f = ecs.poll(nanos, TimeUnit.NANOSECONDS);//等待 if (f == null)//返回空,超時(shí)拋出異常,結(jié)束 throw new TimeoutException(); nanos = deadline - System.nanoTime();//剩余時(shí)間 } else f = ecs.take();//阻塞等待獲取 } if (f != null) {//說(shuō)明已經(jīng)執(zhí)行完 --active;//任務(wù)數(shù)-1 try { return f.get();//返回執(zhí)行結(jié)果 } catch (ExecutionException eex) { ee = eex; } catch (RuntimeException rex) { ee = new ExecutionException(rex); } } } if (ee == null) ee = new ExecutionException(); throw ee; } finally { //取消其他任務(wù),畢竟第一個(gè)結(jié)果已經(jīng)返回了 for (int i = 0, size = futures.size(); i < size; i++) futures.get(i).cancel(true); } } public T invokeAny(Collection extends Callable > tasks) throws InterruptedException, ExecutionException { try { return doInvokeAny(tasks, false, 0); } catch (TimeoutException cannotHappen) { assert false; return null; } } public T invokeAny(Collection extends Callable > tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { return doInvokeAny(tasks, true, unit.toNanos(timeout)); }
返回所有任務(wù)的結(jié)果
publicList > invokeAll(Collection extends Callable > tasks) throws InterruptedException { if (tasks == null) throw new NullPointerException(); ArrayList > futures = new ArrayList >(tasks.size());// boolean done = false; try { for (Callable t : tasks) {//封裝任務(wù),并提交 RunnableFuture f = newTaskFor(t); futures.add(f); execute(f); } for (int i = 0, size = futures.size(); i < size; i++) { Future f = futures.get(i); if (!f.isDone()) { try { f.get();//阻塞,等待結(jié)果 } catch (CancellationException ignore) { } catch (ExecutionException ignore) { } } } done = true; return futures; } finally { if (!done)//有異常,取消 for (int i = 0, size = futures.size(); i < size; i++) futures.get(i).cancel(true); } } public List > invokeAll(Collection extends Callable > tasks, long timeout, TimeUnit unit) throws InterruptedException { if (tasks == null) throw new NullPointerException(); long nanos = unit.toNanos(timeout); ArrayList > futures = new ArrayList >(tasks.size()); boolean done = false; try { for (Callable t : tasks) futures.add(newTaskFor(t)); final long deadline = System.nanoTime() + nanos; final int size = futures.size(); // Interleave time checks and calls to execute in case // executor doesn"t have any/much parallelism. for (int i = 0; i < size; i++) { execute((Runnable)futures.get(i)); nanos = deadline - System.nanoTime(); if (nanos <= 0L) return futures;//每個(gè)提交都要判斷,超時(shí)了返回Future } for (int i = 0; i < size; i++) { Future f = futures.get(i); if (!f.isDone()) { if (nanos <= 0L) return futures; try { f.get(nanos, TimeUnit.NANOSECONDS); } catch (CancellationException ignore) { } catch (ExecutionException ignore) { } catch (TimeoutException toe) { return futures; } nanos = deadline - System.nanoTime(); } } done = true; return futures; } finally { if (!done) for (int i = 0, size = futures.size(); i < size; i++) futures.get(i).cancel(true); } }
文章版權(quán)歸作者所有,未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為,您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請(qǐng)注明本文地址:http://m.hztianpu.com/yun/75708.html
摘要:系統(tǒng)預(yù)定了幾個(gè)線(xiàn)程池,不過(guò)建議手動(dòng)創(chuàng)建,以防止錯(cuò)誤創(chuàng)建消耗資源,比如創(chuàng)建太多線(xiàn)程或者固定線(xiàn)程數(shù)量,無(wú)界隊(duì)列固定線(xiàn)程數(shù)量,數(shù)量為,無(wú)界隊(duì)列,會(huì)按順序執(zhí)行不限制線(xiàn)程數(shù)量,使用隊(duì)列,使用于短任務(wù)基于用于周期性執(zhí)行任務(wù)示例第一個(gè)是,第二個(gè)是第一 系統(tǒng)預(yù)定了幾個(gè)線(xiàn)程池,不過(guò)建議手動(dòng)創(chuàng)建,以防止錯(cuò)誤創(chuàng)建消耗資源,比如創(chuàng)建太多線(xiàn)程或者OOM FixedThreadPool 固定線(xiàn)程數(shù)量,無(wú)界隊(duì)列 p...
摘要:是所有線(xiàn)程池實(shí)現(xiàn)的父類(lèi),我們先看看構(gòu)造函數(shù)構(gòu)造參數(shù)線(xiàn)程核心數(shù)最大線(xiàn)程數(shù)線(xiàn)程空閑后,存活的時(shí)間,只有線(xiàn)程數(shù)大于的時(shí)候生效存活時(shí)間的單位任務(wù)的阻塞隊(duì)列創(chuàng)建線(xiàn)程的工程,給線(xiàn)程起名字當(dāng)線(xiàn)程池滿(mǎn)了,選擇新加入的任務(wù)應(yīng)該使用什么策略,比如拋異常丟棄當(dāng)前 ThreadPoolExecutor ThreadPoolExecutor是所有線(xiàn)程池實(shí)現(xiàn)的父類(lèi),我們先看看構(gòu)造函數(shù) 構(gòu)造參數(shù) corePool...
摘要:接口用于提交任務(wù)接口繼承了接口設(shè)置線(xiàn)程的狀態(tài),還沒(méi)執(zhí)行的線(xiàn)程會(huì)被中斷設(shè)置線(xiàn)程的狀態(tài),嘗試停止正在進(jìn)行的線(xiàn)程當(dāng)調(diào)用或方法后返回為當(dāng)調(diào)用方法后,并且所有提交的任務(wù)完成后返回為當(dāng)調(diào)用方法后,成功停止后返回為當(dāng)前線(xiàn)程阻塞,直到線(xiàn)程執(zhí)行完時(shí)間到被中斷 Executor接口 void execute(Runnable command)//用于提交command任務(wù) ExecutorService接...
摘要:與執(zhí)行方法,是用來(lái)啟動(dòng)線(xiàn)程的,此時(shí)線(xiàn)程處于就緒狀態(tài),獲得調(diào)度后運(yùn)行方法。執(zhí)行方法,相對(duì)于普通方法調(diào)用,在主線(xiàn)程調(diào)用。程序是順序執(zhí)行的,執(zhí)行完才會(huì)執(zhí)行下面的程序。 start與run 執(zhí)行start方法,是用來(lái)啟動(dòng)線(xiàn)程的,此時(shí)線(xiàn)程處于就緒狀態(tài),獲得調(diào)度后運(yùn)行run方法。run方法執(zhí)行結(jié)束,線(xiàn)程就結(jié)束。 執(zhí)行run方法,相對(duì)于普通方法調(diào)用,在主線(xiàn)程調(diào)用。程序是順序執(zhí)行的,執(zhí)行完才會(huì)執(zhí)行下...
摘要:一使用線(xiàn)程池的好處線(xiàn)程池提供了一種限制和管理資源包括執(zhí)行一個(gè)任務(wù)。每個(gè)線(xiàn)程池還維護(hù)一些基本統(tǒng)計(jì)信息,例如已完成任務(wù)的數(shù)量。通過(guò)重復(fù)利用已創(chuàng)建的線(xiàn)程降低線(xiàn)程創(chuàng)建和銷(xiāo)毀造成的消耗。使用無(wú)界隊(duì)列作為線(xiàn)程池的工作隊(duì)列會(huì)對(duì)線(xiàn)程池帶來(lái)的影響與相同。 歷史優(yōu)質(zhì)文章推薦: Java并發(fā)編程指南專(zhuān)欄 分布式系統(tǒng)的經(jīng)典基礎(chǔ)理論 可能是最漂亮的Spring事務(wù)管理詳解 面試中關(guān)于Java虛擬機(jī)(jvm)的問(wèn)...
閱讀 1516·2021-11-11 16:54
閱讀 9862·2021-11-02 14:44
閱讀 2440·2021-10-22 09:53
閱讀 3323·2019-08-30 11:18
閱讀 2010·2019-08-29 13:29
閱讀 2076·2019-08-27 10:58
閱讀 1703·2019-08-26 11:38
閱讀 3586·2019-08-26 10:31