摘要:抽象類,實現了的接口。將任務封裝成提交任務主要方法在任務是否超時超時時間任務書用于存放結果的,先完成的放前面。
AbstractExecutorService抽象類,實現了ExecutorService的接口。
newTaskFor將任務封裝成FutureTask
protectedsubmitRunnableFuture newTaskFor(Runnable runnable, T value) { return new FutureTask (runnable, value); } protected RunnableFuture newTaskFor(Callable callable) { return new FutureTask (callable); }
提交任務
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任務 //timed是否超時 //nanos超時時間 privateinvokeAllT doInvokeAny(Collection extends Callable > tasks, boolean timed, long nanos) throws InterruptedException, ExecutionException, TimeoutException { if (tasks == null) throw new NullPointerException(); int ntasks = tasks.size();//任務書 if (ntasks == 0) throw new IllegalArgumentException(); ArrayList > futures = new ArrayList >(ntasks); //用于存放結果的,先完成的放前面。所以第一個任務沒完成的時候,會繼續提交后續任務 ExecutorCompletionService ecs = new ExecutorCompletionService (this); try { //異常信息 ExecutionException ee = null; //過期時間 final long deadline = timed ? System.nanoTime() + nanos : 0L; Iterator extends Callable > it = tasks.iterator();//獲取第一個任務 提交任務 futures.add(ecs.submit(it.next())); --ntasks;//因為提交了一個,任務數-1 int active = 1;//正在執行的任務 for (;;) { Future f = ecs.poll(); if (f == null) {//第一個沒完成 if (ntasks > 0) {//還有沒提交的任務 --ntasks;//任務數-1 futures.add(ecs.submit(it.next()));//提交任務 ++active;//正在執行的任務+1 } else if (active == 0)//當前沒任務了,但是都失敗了,異常被捕獲了 break; else if (timed) { f = ecs.poll(nanos, TimeUnit.NANOSECONDS);//等待 if (f == null)//返回空,超時拋出異常,結束 throw new TimeoutException(); nanos = deadline - System.nanoTime();//剩余時間 } else f = ecs.take();//阻塞等待獲取 } if (f != null) {//說明已經執行完 --active;//任務數-1 try { return f.get();//返回執行結果 } catch (ExecutionException eex) { ee = eex; } catch (RuntimeException rex) { ee = new ExecutionException(rex); } } } if (ee == null) ee = new ExecutionException(); throw ee; } finally { //取消其他任務,畢竟第一個結果已經返回了 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)); }
返回所有任務的結果
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) {//封裝任務,并提交 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();//阻塞,等待結果 } 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;//每個提交都要判斷,超時了返回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); } }
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/75708.html
摘要:系統預定了幾個線程池,不過建議手動創建,以防止錯誤創建消耗資源,比如創建太多線程或者固定線程數量,無界隊列固定線程數量,數量為,無界隊列,會按順序執行不限制線程數量,使用隊列,使用于短任務基于用于周期性執行任務示例第一個是,第二個是第一 系統預定了幾個線程池,不過建議手動創建,以防止錯誤創建消耗資源,比如創建太多線程或者OOM FixedThreadPool 固定線程數量,無界隊列 p...
摘要:是所有線程池實現的父類,我們先看看構造函數構造參數線程核心數最大線程數線程空閑后,存活的時間,只有線程數大于的時候生效存活時間的單位任務的阻塞隊列創建線程的工程,給線程起名字當線程池滿了,選擇新加入的任務應該使用什么策略,比如拋異常丟棄當前 ThreadPoolExecutor ThreadPoolExecutor是所有線程池實現的父類,我們先看看構造函數 構造參數 corePool...
摘要:接口用于提交任務接口繼承了接口設置線程的狀態,還沒執行的線程會被中斷設置線程的狀態,嘗試停止正在進行的線程當調用或方法后返回為當調用方法后,并且所有提交的任務完成后返回為當調用方法后,成功停止后返回為當前線程阻塞,直到線程執行完時間到被中斷 Executor接口 void execute(Runnable command)//用于提交command任務 ExecutorService接...
摘要:與執行方法,是用來啟動線程的,此時線程處于就緒狀態,獲得調度后運行方法。執行方法,相對于普通方法調用,在主線程調用。程序是順序執行的,執行完才會執行下面的程序。 start與run 執行start方法,是用來啟動線程的,此時線程處于就緒狀態,獲得調度后運行run方法。run方法執行結束,線程就結束。 執行run方法,相對于普通方法調用,在主線程調用。程序是順序執行的,執行完才會執行下...
摘要:一使用線程池的好處線程池提供了一種限制和管理資源包括執行一個任務。每個線程池還維護一些基本統計信息,例如已完成任務的數量。通過重復利用已創建的線程降低線程創建和銷毀造成的消耗。使用無界隊列作為線程池的工作隊列會對線程池帶來的影響與相同。 歷史優質文章推薦: Java并發編程指南專欄 分布式系統的經典基礎理論 可能是最漂亮的Spring事務管理詳解 面試中關于Java虛擬機(jvm)的問...
閱讀 1439·2021-11-11 16:54
閱讀 9319·2021-11-02 14:44
閱讀 2371·2021-10-22 09:53
閱讀 3259·2019-08-30 11:18
閱讀 1951·2019-08-29 13:29
閱讀 2003·2019-08-27 10:58
閱讀 1623·2019-08-26 11:38
閱讀 3518·2019-08-26 10:31