CloseableExecutorService

對Old ExecutorService進行了裝飾以創(chuàng)建一個可關(guān)閉Running的Future,Runnable的New ExecutorService

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Decoration on an ExecutorService that tracks created futures and provides
 * a method to close futures created via this class
 */
public class CloseableExecutorService implements Closeable
{
    private final Logger log = LoggerFactory.getLogger(CloseableExecutorService.class);
    private final Set<Future<?>> futures = Sets.newSetFromMap(Maps.<Future<?>, Boolean>newConcurrentMap());
    private final ExecutorService executorService;
    private final boolean shutdownOnClose;
    protected final AtomicBoolean isOpen = new AtomicBoolean(true);

    protected class InternalScheduledFutureTask implements Future<Void>
    {
        private final ScheduledFuture<?> scheduledFuture;

        public InternalScheduledFutureTask(ScheduledFuture<?> scheduledFuture)
        {
            this.scheduledFuture = scheduledFuture;
            futures.add(scheduledFuture);
        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning)
        {
            futures.remove(scheduledFuture);
            return scheduledFuture.cancel(mayInterruptIfRunning);
        }

        @Override
        public boolean isCancelled()
        {
            return scheduledFuture.isCancelled();
        }

        @Override
        public boolean isDone()
        {
            return scheduledFuture.isDone();
        }

        @Override
        public Void get() throws InterruptedException, ExecutionException
        {
            return null;
        }

        @Override
        public Void get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
        {
            return null;
        }
    }

    protected class InternalFutureTask<T> extends FutureTask<T>
    {
        private final RunnableFuture<T> task;

        InternalFutureTask(RunnableFuture<T> task)
        {
            super(task, null);
            this.task = task;
            futures.add(task);
        }

        protected void done()
        {
            futures.remove(task);
        }
    }

    /**
     * @param executorService the service to decorate
     */
    public CloseableExecutorService(ExecutorService executorService)
    {
       this(executorService, false);
    }

    /**
     * @param executorService the service to decorate
     * @param shutdownOnClose if true, shutdown the executor service when this is closed
     */
    public CloseableExecutorService(ExecutorService executorService, boolean shutdownOnClose)
    {
        this.executorService = Preconditions.checkNotNull(executorService, "executorService cannot be null");
        this.shutdownOnClose = shutdownOnClose;
    }

    /**
     * Returns <tt>true</tt> if this executor has been shut down.
     *
     * @return <tt>true</tt> if this executor has been shut down
     */
    public boolean isShutdown()
    {
        return !isOpen.get();
    }

    @VisibleForTesting
    int size()
    {
        return futures.size();
    }

    /**
     * Closes any tasks currently in progress
     */
    @Override
    public void close()
    {
        isOpen.set(false);
        Iterator<Future<?>> iterator = futures.iterator();
        while ( iterator.hasNext() )
        {
            Future<?> future = iterator.next();
            iterator.remove();
            if ( !future.isDone() && !future.isCancelled() && !future.cancel(true) )
            {
                log.warn("Could not cancel " + future);
            }
        }
        if (shutdownOnClose) {
            executorService.shutdownNow();
        }
    }

    /**
     * Submits a value-returning task for execution and returns a Future
     * representing the pending results of the task.  Upon completion,
     * this task may be taken or polled.
     *
     * @param task the task to submit
     * @return a future to watch the task
     */
    public<V> Future<V> submit(Callable<V> task)
    {
        Preconditions.checkState(isOpen.get(), "CloseableExecutorService is closed");

        InternalFutureTask<V> futureTask = new InternalFutureTask<V>(new FutureTask<V>(task));
        executorService.execute(futureTask);
        return futureTask;
    }

    /**
     * Submits a Runnable task for execution and returns a Future
     * representing that task.  Upon completion, this task may be
     * taken or polled.
     *
     * @param task the task to submit
     * @return a future to watch the task
     */
    public Future<?> submit(Runnable task)
    {
        Preconditions.checkState(isOpen.get(), "CloseableExecutorService is closed");

        InternalFutureTask<Void> futureTask = new InternalFutureTask<Void>(new FutureTask<Void>(task, null));
        executorService.execute(futureTask);
        return futureTask;
    }
}
最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點,簡書系信息發(fā)布平臺,僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • Spring Cloud為開發(fā)人員提供了快速構(gòu)建分布式系統(tǒng)中一些常見模式的工具(例如配置管理,服務(wù)發(fā)現(xiàn),斷路器,智...
    卡卡羅2017閱讀 136,711評論 19 139
  • 個人筆記,方便自己查閱使用 Contents Java LangAssignment, ReferenceData...
    freenik閱讀 1,541評論 0 6
  • 1. Java基礎(chǔ)部分 基礎(chǔ)部分的順序:基本語法,類相關(guān)的語法,內(nèi)部類的語法,繼承相關(guān)的語法,異常的語法,線程的語...
    子非魚_t_閱讀 34,854評論 18 399
  • 從三月份找實習(xí)到現(xiàn)在,面了一些公司,掛了不少,但最終還是拿到小米、百度、阿里、京東、新浪、CVTE、樂視家的研發(fā)崗...
    時芥藍閱讀 42,890評論 11 349
  • 只要一有空余時間她就到花店幫忙,跟老師學(xué)花藝。她已深深愛上了這里。平時天天都在上班,做著自己不喜歡的工作,只為...
    日落西斜閱讀 950評論 0 4

友情鏈接更多精彩內(nèi)容