如何例用Executor创建后台线程池

Executor

执行已提交的Runnable任务的对象接口。这个接口提供了一种将任务提交和任务运行机制（包括线程使用、调度等的详细信息）分离的方法。

public interface Executor {

    /**
     * 在将来的某些时候执行给定任务。根据Executor的实现判定是在一个新线程还是线程池还是调用线 
     * 程中执行。
     */
    void execute(Runnable command);
}

通常用Executor来代替显示的创建Thread。如要在单独的线程执行某项任务时，使用

    public void test() {
        Executor executor = new ThreadPerTaskExecutor();
        executor.execute(new RunnableTask1());
    }

    class ThreadPerTaskExecutor implements Executor {
        public void execute(Runnable r) {
            new Thread(r).start();
        }
    }

而不是

new Thread(new RunnableTask()).start()

很多Executor的实现都对任务的执行方式和时间强加的限制，如下面定义了一个顺序执行任务的Executor：

class SerialExecutor implements Executor {
        final Queue<Runnable> tasks = new ArrayDeque<>();
        final Executor executor;
        Runnable active;

        SerialExecutor(Executor executor) {
            this.executor = executor;
        }

        public synchronized void execute(final Runnable r) {
            tasks.add(() -> {
                try {
                    r.run();
                } finally {
                    scheduleNext();
                }
            });
            if (active == null) {
                scheduleNext();
            }
        }

        protected synchronized void scheduleNext() {
            if ((active = tasks.poll()) != null) {
                executor.execute(active);
            }
        }
    }

java.util.concurrent包中提供的Executor的实现ExecutorService是一个更广泛的被使用的接口。ThreadPoolExecutor提供一个可扩展的线程池实现。Executors类为这些Executor提供了方便使用的工厂方法。

下面为java.util.concurrent包中Executor框架的主要类关系图：
Executor框架.png

ExecutorService

  1. 一个Executor；
  2. 提供shutdown()方法终止Executor的执行；
  3. 提供submit()方法返回一个Future来追踪异步任务的执行进度；
  4. 提供invokeAll()方法返回一个Future列表来追踪多个异步任务的执行进度。

如果我们现在创建了一个异步执行的网络任务：

    class NetworkService implements Runnable {
        private final ServerSocket serverSocket;
        private final ExecutorService pool;

        public NetworkService(int port, int poolSize)
                throws IOException {
            serverSocket = new ServerSocket(port);
            pool = Executors.newFixedThreadPool(poolSize);
        }

        public void run() { // run the service
            try {
                for (; ; ) {
                    pool.execute(new Handler(serverSocket.accept()));
                }
            } catch (IOException ex) {
                pool.shutdown();
            }
        }
    }

    class Handler implements Runnable {
        private final Socket socket;

        Handler(Socket socket) {
            this.socket = socket;
        }

        public void run() {
            // read and service request on socket
        }
    }

可以调用shutdown()方法终止它的执行：

    void shutdownAndAwaitTermination(ExecutorService pool) {
        pool.shutdown(); // Disable new tasks from being submitted
        try {
            // Wait a while for existing tasks to terminate
            if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
                pool.shutdownNow(); // Cancel currently executing tasks
                // Wait a while for tasks to respond to being cancelled
                if (!pool.awaitTermination(60, TimeUnit.SECONDS))
                    System.err.println("Pool did not terminate");
            }
        } catch (InterruptedException ie) {
            // (Re-)Cancel if current thread also interrupted
            pool.shutdownNow();
            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }
    }
ScheduledExecutorService
  1. 一个ExecutorService；
  2. 其scheduleXXX()方法创建并执行ScheduledFuture；
  3. 该ScheduledFuture可以在给定的延迟后变为启动状态，并开始执行任务；
  4. 该ScheduledFuture可以定期执行任务；
  5. 通过这个ScheduledFuture可以get任务执行结果，可以cancel这个任务。

如需要创建一个在一小时内每隔10s就响一次的报警器：

    class BeeperControl {
        private final ScheduledExecutorService scheduler =
                Executors.newScheduledThreadPool(1);

        public void beepForAnHour() {
            final Runnable beeper = () -> System.out.println("beep");

            final ScheduledFuture<?> beeperHandle =
                    scheduler.scheduleAtFixedRate(beeper, 10, 10, SECONDS);

            scheduler.schedule(() -> {
                beeperHandle.cancel(true);
            }, 60 * 60, SECONDS);
        }
    }
AbstractExecutorService
  1. 一个ExecutorService；
  2. 其submit、invokeAny、invokeAll方法在执行任务前，把任务封装成RunnableFuture，方便跟踪执行进度。
    /**
     * Returns a {@code RunnableFuture} for the given runnable and default
     * value.
     *
     * @param runnable the runnable task being wrapped
     * @param value the default value for the returned future
     * @param <T> the type of the given value
     * @return a {@code RunnableFuture} which, when run, will run the
     * underlying runnable and which, as a {@code Future}, will yield
     * the given value as its result and provide for cancellation of
     * the underlying task
     * @since 1.6
     */
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        return new FutureTask<T>(runnable, value);
    }

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

ExecutorService的实现

ThreadPoolExecutor

  1. 一个AbstractExecutorService；
  2. 使用线程池中的某个线程执行提交的任务，该线程池通常由Executors中的工厂方法创建；
  3. FIFO；
  4. 其execute的设计思想为：
if(threadCount < 池中需要在空闲状态也要保持的线程数){
      new thread;
      把当前任务做为新建线程的第一个任务执行；
} else if(任务放入等待队列){
     在队列中等待在运行的线程将它取出并执行；
} else if(threadCount < 池中最多可拥有的线程数){ //此时意味着等待队列已满，放入失败
     new thread;
     把当前任务做为新建线程的第一个任务执行；
} else {
     拒绝该任务的执行，如抛出异常；
}

  1. 线程池中维持corePoolSize数量的线程数，如果有线程在线程池shutdown之前因为执行失败而中断，则在执行后续任务时会新建一个替代它的线程。

    具体原理如下：
    ThreadPoolExecutor.png

其中定义了线程池的control state(ctl)：
ThreadPoolExecutor线程池状态.png

各个ctl之间的关系如下：
ThreadPoolExecutor runState.png

所以shutdown()代表停止接收新任务但仍需执行现有任务，shutdownNow()代表停止接收新任务也不再执行现有任务。

ThreadPoolExecutor在实现时用到了BlockQueue和AbstracQueuedSynchronizer。

BlockQueue比较常见的实现为LinkedBlockingQueue。其原理如下：
LinkedBlockingQueue.png
AbstractQueuedSynchronizer.png

可以看到Lock和BlockQueue是非常实用的。
Lock替代了synchronized，功能更强大使用也更简便，如下：

    class BoundedBuffer {
        final Lock lock = new ReentrantLock();
        final Condition notFull = lock.newCondition();
        final Condition notEmpty = lock.newCondition();

        final Object[] items = new Object[100];
        int putptr, takeptr, count;

        public void put(Object x) throws InterruptedException {
            lock.lock();
            try {
                while (count == items.length)
                    notFull.await();
                items[putptr] = x;
                if (++putptr == items.length) putptr = 0;
                ++count;
                notEmpty.signal();
            } finally {
                lock.unlock();
            }
        }

        public Object take() throws InterruptedException {
            lock.lock();
            try {
                while (count == 0)
                    notEmpty.await();
                Object x = items[takeptr];
                if (++takeptr == items.length) takeptr = 0;
                --count;
                notFull.signal();
                return x;
            } finally {
                lock.unlock();
            }
        }
    }

BlockingQueue非常适用于生产-消费者模型，如下：

class Producer implements Runnable {
    private final BlockingQueue queue;
    Producer(BlockingQueue q) { queue = q; }
    public void run() {
        try {
            while (true) { queue.put(produce()); }
        } catch (InterruptedException ex) { ... handle ...}
    }
    Object produce() { ... }
}

class Consumer implements Runnable {
    private final BlockingQueue queue;

    Consumer(BlockingQueue q) {
        queue = q;
    }

    public void run() {
        try {
            while (true) {
                consume(queue.take());
            }
        } catch (InterruptedException ex) { ...handle ...}
    }

    void consume(Object x) { ...}
}

class BlockingQueueTest {
    void main() {
        BlockingQueue q = new SomeQueueImplementation();
        LockTest.Producer p = new LockTest.Producer(q);
        Consumer c1 = new Consumer(q);
        Consumer c2 = new Consumer(q);
        new Thread(p).start();
        new Thread(c1).start();
        new Thread(c2).start();
    }
}

ScheduledThreadPoolExecutor

  1. 一个可以执行延迟任务和周期性任务的ThreadPoolExecutor；
  2. 如果提交的任务在运行前被取消，则该任务会在延迟结束后被取消；
  3. 如果多个任务有相同的执行时间，采用FIFO；

因为功能不同，其实现和ThreadPoolExecutor也不一样：

  1. maximumPoolSize == corePoolSize；且等待任务队列大小没有限制为Integer.MAX_VALUE；
  2. 设置了shutdown策略来决定哪些任务在线程池进入SHUTDOWN状态后不需要再执行；
  3. 采用DelayedWorkQueue而非LinkedBlockingQueue，加快取消速度（时间复杂度从O(n)降至O(log n)）；
  4. 在插入队列时将所有任务封装成ScheduledFutureTask给DelayedWorkQueue使用；

具体实现原理如下：
ScheduledThreadPoolExecutor.png

ForkJoinPool

  1. 继承AbstractExecutorService；
  2. 将所有提交的或执行的任务封装成ForkJoinTask执行；
  3. 需要将任务分解成足够小的子任务执行，思路如下：
解决(问题):
    if 问题足够小:
        直接解决问题 (顺序算法)
    else:
        for 部份 in 细分(问题)
            fork 子任务来解决(部份)
        join 在前面的循环中生成的所有子任务
        return 合并的结果
fork-join model.png
  1. 每个Worker都有其自己的本地任务队列；

  2. 每个Worker都可以去其它Worker的任务队列中窃取任务执行；


    work stealing实现方式.png
  3. workQueues保存了WorkQueue列表，其中奇数索引代表内部任务（unshared queues，本地任务列表），偶数索引代表外部任务（shared queues，窃取的任务列表），因此workQueues的大小始终为2的幂次方。
ForkJoinPool workQueues.png

其具体实现原理如下：


ForkJoinPool.png

Executors

提供java.util.concurrent包中的Executor、ExecutorService、ScheduledExecutorService、ThreadFactory的工厂方法。

newFixedThreadPool()

创建一个指定corePoolSize和maximumPoolSize都为nThreads的ThreadPoolExecutor。

适用于负载较重的并行运算。

newWorkStealingPool()

创建ForkJoinPool，指定处理还没joined的forked任务的规则为FIFO（上面介绍中的ForkJoinPool默认采用LIFO）。

适用于“分而治之”递归运算计算密集的运算。

newScheduledThreadPool()

创建一个指定corePoolSize的ScheduledThreadPoolExecutor。

适用于需要定期或周期性执行的运算。

newSingleThreadExecutor()

创建一个仅仅只能拥有一个执行线程的ThreadPoolExecutor。

适用于串行运算。

newCachedThreadPool()

创建一个线程数量不限的ThreadPoolExecutor（corePoolSize=0, maximumPoolSize=Integer.MAX_VALUE）。

适用于负载较轻的数量繁多的短期运算。

newSingleThreadScheduledExecutor()

创建一个仅仅只能拥有一个执行线程的ScheduledThreadPoolExecutor。

适用于串行的需要定期或周期性执行的运算。

MoreExecutors

提供java.util.concurrent包中的Executor、ExecutorService、ThreadFactory的工厂方法。

newSequentialExecutor(Executor delegate)

delegate：实际执行任务的底层Executor

创建一个实际由delegate执行任务的SequentialExecutor。

SequentialExecutor：由delegate委拖任务执行程序实际执行任务的采用FIFO规则串行执行任务的Executor。

适用于需要按FIFO规则执行的串行运算。

SequentialExecutor.png

listeningDecorator(ExecutorService delegate)

delegate：实际执行任务的底层Executor

创建一个实际由delegate执行任务的ListeningExecutorService。

ListeningExecutorService：可以在任务完成后再执行给定任务的ExecutorService。

适用于需要将运算链接在一起的场景。

ListeningExecutorService.png

应用

创建一个在非UI线程执行并行任务的Executor

    @Provides
    @Annotations.NonUiParallel
    @Singleton
    static ExecutorService provideNonUiThreadPool() {
        return Executors.newFixedThreadPool(
                5,
                runnable -> {
                    Log.i("DialerExecutorModule.newThread", "creating low priority thread");
                    Thread thread = new Thread(runnable, "DialerExecutors-LowPriority");
                    // Java thread priority 4 corresponds to Process.THREAD_PRIORITY_BACKGROUND (10)
                    thread.setPriority(4);
                    return thread;
                });
    }

创建一个在非UI线程执行串行任务的Executor

    static ScheduledExecutorService provideNonUiSerialExecutorService() {
        return Executors.newSingleThreadScheduledExecutor(
                runnable -> {
                    Log.i("NonUiTaskBuilder.newThread", "creating serial thread");
                    Thread thread = new Thread(runnable, "DialerExecutors-LowPriority-Serial");
                    // Java thread priority 4 corresponds to Process.THREAD_PRIORITY_BACKGROUND (10)
                    thread.setPriority(4);
                    return thread;
                });
    }

创建一个在UI线程执行延迟或周期性任务的Executor

    static ScheduledExecutorService provideUiSerialExecutorService() {
        return Executors.newSingleThreadScheduledExecutor(
                runnable -> {
                    Log.i("DialerExecutorModule.newThread", "creating serial thread");
                    Thread thread = new Thread(runnable, "DialerExecutors-HighPriority-Serial");
                    // Java thread priority 5 corresponds to Process.THREAD_PRIORITY_DEFAULT (0)
                    thread.setPriority(5);
                    return thread;
                });
    }

创建一个在UI线程执行的可以注册listener的轻量级Executor

    @Provides
    @Annotations.UiParallel
    @Singleton
    static ExecutorService provideUiThreadPool() {
        return (ExecutorService) AsyncTask.THREAD_POOL_EXECUTOR;
    }

    static ListeningExecutorService provideLightweightExecutor(@Annotations.UiParallel ExecutorService delegate) {
        return MoreExecutors.listeningDecorator(delegate);
    }

创建一个在非UI线程异步执行运算的可以注册listener的Executore

    static ListeningExecutorService provideBackgroundExecutor(
            @Annotations.NonUiParallel ExecutorService delegate) {
        return MoreExecutors.listeningDecorator(delegate);
    }

总结

Executor框架把任务的提交和执行解耦，可以简化并发编程。线程能过线程的复用，可以降低资源消耗，提高响应速度等。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 216,651评论 6赞 501
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 92,468评论 3赞 392
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 162,931评论 0赞 353
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 58,218评论 1赞 292
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 67,234评论 6赞 388
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 51,198评论 1赞 299
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,084评论 3赞 418
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 38,926评论 0赞 274
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 45,341评论 1赞 311
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,563评论 2赞 333
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,731评论 1赞 348
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 35,430评论 5赞 343
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,036评论 3赞 326
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,676评论 0赞 22
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,829评论 1赞 269
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 47,743评论 2赞 368
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 44,629评论 2赞 354