麻豆小视频在线观看_中文黄色一级片_久久久成人精品_成片免费观看视频大全_午夜精品久久久久久久99热浪潮_成人一区二区三区四区

首頁 > 學院 > 開發設計 > 正文

《java.util.concurrent 包源碼閱讀》10 線程池系列之AbstractExecutorService

2019-11-14 20:52:12
字體:
來源:轉載
供稿:網友
java.util.concurrent 包源碼閱讀》10 線程池系列之AbstractExecutorService

AbstractExecutorService對ExecutorService的執行任務類型的方法提供了一個默認實現。這些方法包括submit,invokeAny和InvokeAll。

注意的是來自Executor接口的execute方法是未被實現,execute方法是整個體系的核心,所有的任務都是在這個方法里被真正執行的,因此該方法的不同實現會帶來不同的執行策略。這個在后面分析ThreadPoolExecutor和ScheduledThreadPoolExecutor就能看出來。

首先來看submit方法,它的基本邏輯是這樣的:

1. 生成一個任務類型和Future接口的包裝接口RunnableFuture的對象

2. 執行任務

3. 返回future。

    public Future<?> submit(Runnable task) {        if (task == null) throw new NullPointerException();        RunnableFuture<Void> ftask = newTaskFor(task, null);        execute(ftask);        return ftask;    }    public <T> Future<T> submit(Callable<T> task) {        if (task == null) throw new NullPointerException();        RunnableFuture<T> ftask = newTaskFor(task);        execute(ftask);        return ftask;    }

因為submit支持Callable和Runnable兩種類型的任務,因此newTaskFor方法有兩個重載方法:

    PRotected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {        return new FutureTask<T>(callable);    }    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {        return new FutureTask<T>(runnable, value);    }

上一篇文章里曾經說過Callable和Runnable的區別在于前者帶返回值,也就是說Callable=Runnable+返回值。因此java中提供了一種adapter,把Runnable+返回值轉換成Callable類型。這點可以在newTaskFor中的FutureTask類型的構造函數的代碼中看到:

    public FutureTask(Callable<V> callable) {        if (callable == null)            throw new NullPointerException();        sync = new Sync(callable);    }    public FutureTask(Runnable runnable, V result) {        sync = new Sync(Executors.callable(runnable, result));    }

以下是Executors.callable方法的代碼:

    public static <T> Callable<T> callable(Runnable task, T result) {        if (task == null)            throw new NullPointerException();        return new RunnableAdapter<T>(task, result);    }

那么RunnableAdapter的代碼就很好理解了,它是一個Callable的實現,call方法的實現就是執行Runnable的run方法,然后返回那個value。

    static final class RunnableAdapter<T> implements Callable<T> {        final Runnable task;        final T result;        RunnableAdapter(Runnable task, T result) {            this.task = task;            this.result = result;        }        public T call() {            task.run();            return result;        }    }

接下來先說說較為簡單的invokeAll:

1. 為每個task調用newTaskFor方法生成得到一個既是Task也是Future的包裝類對象的List

2. 循環調用execute執行每個任務

3. 再次循環調用每個Future的get方法等待每個task執行完成

4. 最后返回Future的list。

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,                                         long timeout, TimeUnit unit)        throws InterruptedException {        if (tasks == null || unit == null)            throw new NullPointerException();        long nanos = unit.toNanos(timeout);        List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());        boolean done = false;        try {            // 為每個task生成包裝對象            for (Callable<T> t : tasks)                futures.add(newTaskFor(t));            long lastTime = System.nanoTime();            // 循環調用execute執行每個方法            // 這里因為設置了超時時間,所以每次執行完成后            // 檢查是否超時,超時了就直接返回future集合            Iterator<Future<T>> it = futures.iterator();            while (it.hasNext()) {                execute((Runnable)(it.next()));                long now = System.nanoTime();                nanos -= now - lastTime;                lastTime = now;                if (nanos <= 0)                    return futures;            }            // 等待每個任務執行完成            for (Future<T> f : futures) {                if (!f.isDone()) {                    if (nanos <= 0)                        return futures;                    try {                        f.get(nanos, TimeUnit.NANOSECONDS);                    } catch (CancellationException ignore) {                    } catch (ExecutionException ignore) {                    } catch (TimeoutException toe) {                        return futures;                    }                    long now = System.nanoTime();                    nanos -= now - lastTime;                    lastTime = now;                }            }            done = true;            return futures;        } finally {            if (!done)                for (Future<T> f : futures)                    f.cancel(true);        }    }

最后說說invokeAny,它的難點在于只要一個任務執行成功就要返回,并且會取消其他任務,也就是說重點在于找到第一個執行成功的任務。

這里我想到了BlockingQueue,當所有的任務被提交后,任務執行返回的Future會被依次添加到一個BlockingQueue中,然后找到第一個執行成功任務的方法就是從BlockingQueue取出第一個元素,這個就是doInvokeAny方法用到的ExecutorCompletionService的基本原理。

因為兩個invokeAny方法都是調用doInvokeAny方法,下面是doInvokeAny的代碼分析:

    private <T> T doInvokeAny(Collection<? extends Callable<T>> 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();        List<Future<T>> futures= new ArrayList<Future<T>>(ntasks);        // ExecutorCompletionService負責執行任務,后面調用用poll返回第一個執行結果        ExecutorCompletionService<T> ecs =            new ExecutorCompletionService<T>(this);        // 這里出于效率的考慮,每次提交一個任務之后,就檢查一下有沒有執行完成的任務        try {            ExecutionException ee = null;            long lastTime = timed ? System.nanoTime() : 0;            Iterator<? extends Callable<T>> it = tasks.iterator();            // 先提交一個任務            futures.add(ecs.submit(it.next()));            --ntasks;            int active = 1;            for (;;) {                // 嘗試獲取有沒有執行結果(這個結果是立刻返回的)                Future<T> f = ecs.poll();                // 沒有執行結果                if (f == null) {                    // 如果還有任務沒有被提交執行的,就再提交一個任務                    if (ntasks > 0) {                        --ntasks;                        futures.add(ecs.submit(it.next()));                        ++active;                    }                    // 沒有任務在執行了,而且沒有拿到一個成功的結果。                    else if (active == 0)                        break;                    // 如果設置了超時情況                    else if (timed) {                        // 等待執行結果直到有結果或者超時                        f = ecs.poll(nanos, TimeUnit.NANOSECONDS);                        if (f == null)                            throw new TimeoutException();                        // 這里的更新不可少,因為這個Future可能是執行失敗的情況,那么還需要再次等待下一個結果,超時的設置還是需要用到。                        long now = System.nanoTime();                        nanos -= now - lastTime;                        lastTime = now;                    }                    // 沒有設置超時,并且所有任務都被提交了,則一直等到第一個執行結果出來                    else                        f = ecs.take();                }                // 有返回結果了,嘗試從future中獲取結果,如果失敗了,那么需要接著等待下一個執行結果                if (f != null) {                    --active;                    try {                        return f.get();                    } catch (ExecutionException eex) {                        ee = eex;                    } catch (RuntimeException rex) {                        ee = new ExecutionException(rex);                    }                }            }            // ExecutorCompletionService執行時發生錯誤返回了全是null的future            if (ee == null)                ee = new ExecutionException();            throw ee;        } finally {            // 嘗試取消所有的任務(對于已經完成的任務沒有影響)            for (Future<T> f : futures)                f.cancel(true);        }    }

后面接著分析ThreadPoolExecutor和ScheduledThreadPoolExecutor。


發表評論 共有條評論
用戶名: 密碼:
驗證碼: 匿名發表
主站蜘蛛池模板: 日韩视频在线观看免费视频 | 最新av免费网址 | 狠狠婷婷综合久久久久久妖精 | 久久人添人人爽人人爽人人片av | a视频在线看 | xvideos korean | 91av亚洲 | 久久久久久久久久亚洲精品 | 国产pron| 亚洲一区在线免费视频 | 粉嫩粉嫩一区二区三区在线播放 | www.狠狠操.com | xx53xx| 免费观看一区二区三区视频 | 黄色毛片视频在线观看 | 国产午夜精品久久久久久久蜜臀 | 国产精品久久国产精麻豆96堂 | 神马久久蜜桃 | 99欧美精品 | 综合网日日天干夜夜久久 | 国产一级二级毛片 | 国产精品看片 | 在线观看中文字幕av | 性猛交ⅹxxx乱巴西 在线播放中文 | 成人短视频在线观看 | 亚洲人成网站免费播放 | 欧美人人干 | 91精品免费在线 | 欧美重口另类videos人妖 | 国产精品视频在 | 欧美性生交xxxxx久久久 | 国产精品9191 | 成人免费乱码大片a毛片视频网站 | 蜜桃视频最新网址 | 成人免费观看49www在线观看 | 91久久夜色精品国产网站 | 蜜桃久久一区二区三区 | 久久国产午夜 | 欧美 日韩 国产 成人 | 毛片在线免费播放 | qyl在线视频精品免费观看 |