Java 线程池 ExecutorService

new Thread的弊端

new Thread(new Runnable() {
@Override
public void run() {
// TODO Auto-generated method stub
}
}).start();

new Thread执行一个异步任务的弊端如下：每次new Thread新建对象性能差。线程缺乏统一管理，可能无限制新建线程，相互之间竞争，及可能占用过多系统资源导致死机或oom。缺乏更多功能，如定时执行、定期执行、线程中断。

相比new Thread，Java提供的四种线程池的好处在于：重用存在的线程，减少对象创建、消亡的开销，性能佳。可有效控制最大并发线程数，提高系统资源的使用率，同时避免过多资源竞争，避免堵塞。提供定时执行、定期执行、单线程、并发数控制等功能。

Java 线程池
Java通过Executors提供四种线程池：

(1). newCachedThreadPool

创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。示例代码如下：

        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            final int index = i;
            try {
                Thread.sleep(index * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            cachedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(index);
                }
            });
        }

线程池为无限大，当执行第二个任务时第一个任务已经完成，会复用执行第一个任务的线程，而不用每次新建线程。

(2). newFixedThreadPool

创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。示例代码如下：

        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 10; i++) {
            final int index = i;
            fixedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println(index);
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            });
        }

因为线程池大小为3，每个任务输出index后sleep 2秒，所以每两秒打印3个数字。定长线程池的大小最好根据系统资源进行设置。如Runtime.getRuntime().availableProcessors()。可参考PreloadDataCache。

(3) newScheduledThreadPool

创建一个定长线程池，支持定时及周期性任务执行。延迟执行示例代码如下：

        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
        scheduledThreadPool.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("delay 3 seconds");
            }
        }, 3, TimeUnit.SECONDS);

表示延迟3秒执行。定期执行示例代码如下：

        scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                System.out.println("delay 1 seconds, and excute every 3 seconds");
            }
        }, 1, 3, TimeUnit.SECONDS);

表示延迟1秒后每3秒执行一次。
ScheduledExecutorService比Timer更安全，功能更强大，后面会有一篇单独进行对比。

(4)、newSingleThreadExecutor

创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。示例代码如下：

        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 10; i++) {
            final int index = i;
            singleThreadExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println(index);
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            });
        }

结果依次输出，相当于顺序执行各个任务。现行大多数GUI程序都是单线程的。Android中单线程可用于数据库操作，文件操作，应用批量安装，应用批量删除等不适合并发但可能IO阻塞性及影响UI线程响应的操作。

submit（）与execute（）区别

execute()方法源码如下：

public interface Executor {
    void execute(Runnable command);
}

execute()方法的入参为一个Runnable，返回值为void，这时候我们已经知道了execute()方法的来源以及其定义。接下来，我们来看看，submit()是从ExecutorService接口中的，打开源码，看到了submit()方法：

public interface ExecutorService extends Executor {
　　...
　　 Future submit(Callable task);
 
　　 Future submit(Runnable task, T result);
 
　　Future submit(Runnable task);
　　...
}

可以看出，在ExecutorService接口中，一共有以上三个sumbit()方法，入参可以为Callable，也可以为Runnable，而且方法有返回值Future；

从上面的源码以及讲解可以总结execute()和submit()方法的区别：

　　1. 接收的参数不一样;

　　2. submit()有返回值，而execute()没有；例如，有个validation的task，希望该task执行完后告诉我它的执行结果，是成功还是失败，然后继续下面的操作。

　　3. submit()可以进行Exception处理；例如，如果task里会抛出checked或者unchecked exception，而你又希望外面的调用者能够感知这些exception并做出及时的处理，那么就需要用到submit，通过对Future.get()进行抛出异常的捕获，然后对其进行处理。

execute()方法实际应用实例，在对list数据进行处理，比如需要调用RestFul或webService服务对list其他字段补齐的时候，就用到execute()方法，代码如下：

 
    public List findListVO(List list) {
        List result = Lists.newArrayList();
        final int taskSize = 2;
        CountDownLatch cdl = new CountDownLatch(taskSize);
        //task-1
        executor.execute(()->deal1(list,cdl));
        //task-2
        executor.execute(()->deal2(list,cdl));
        return result;
    }
    
    
    private void deal1(List list,CountDownLatch cdl) {
        try{
            if(list.size()!=0) {
                list.forEach(item->{});
            }
        } finally {
            cdl.countDown();
        }
    }
    
    
    private void deal2(List list,CountDownLatch cdl) {
        try{
            if(list.size()!=0) {
                list.forEach(item->{});
            }
        } finally {
            cdl.countDown();
        }
    }

submit()方法在实际应用中的实例，比如一个RestFul接口，只支持一个参数查询，上游系统无法改变传参，这时，为了节省效率，支持多参数查询，我们不得不循环去调用接口，这会导致我们的性能非常慢，为了节省效率，我们不得不把查询放到异步里面去做。代码如下：

    public Map> findMap(Map map) throws Exception {
        Set set = map.keySet();
        Map> resultMap = Maps.newHashMap();
        if(set.size()==1) {
            Map mapResult = queryMap(set.stream().findFirst().get());
            resultMap.put(set.stream().findFirst().get(),mapResult);
        } else {
            try {
                CountDownLatch cdl = new CountDownLatch(set.size());
                set.forEach(item->{
                    es.submit(()-> {
                        try{
                            Map maps = queryMap(item);
                            resultMap.put(item,maps);
                        } finally {
                            cdl.countDown();
                        }    
                    });    
                });
        } catch(Exception e) {
            throw new Exception();
        }
      }
        return resultMap;
    }
 
   
    public static Map queryMap(String contract) {
        Map map = Maps.newHashMap();
        return map;
    }

shotdown（） showdownNow（）区别

shutdown只是将线程池的状态设置为SHUTWDOWN状态，正在执行的任务会继续执行下去，没有被执行的则中断。而shutdownNow则是将线程池的状态设置为STOP，正在执行的任务则被停止，没被执行任务的则返回。

    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            //将线程池状态设置为shutdown
            advanceRunState(SHUTDOWN);
            //中断线程池中的worker线程
            interruptIdleWorkers();
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

那么中断线程的操作就在这个interruptIdleWorkers()中。

    private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers) {
                Thread t = w.thread;
                //这里的tryLock是尝试获取工作线程的锁，如果工作线程在运行
                //那么就获取不到锁，也就是无法中断。所以这里中断的都是没有运行的线程
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        //调用每个线程的中断
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        w.unlock();
                    }
                }
                if (onlyOne)
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }

所以说shutdown是不会让线程直接退出的，之前添加的线程还是会运行到自然结束。但是因为这块设置了线程池状态为shutdown，因此线程是添加不了的。而在shutdownNow方法中

    public List shutdownNow() {
        List tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            //直接设置为stop状态
            advanceRunState(STOP);
            //中断线程
            interruptWorkers();
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
    }

    private void interruptWorkers() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers)
               //逐个进行中断
                w.interruptIfStarted();
        } finally {
            mainLock.unlock();
        }
    }

    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
            //直接调用中断
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }

所以看的出来，ShutDownNow是直接中断所有的工作线程。并且阻止新线程的添加。

Runnable（）与Callable（）区别
如果是一个多线程协作程序，比如菲波拉切数列，1，1，2，3，5，8…使用多线程来计算。
但后者需要前者的结果，就需要用callable接口了。
callable用法和runnable一样，只不过调用的是call方法，该方法有一个泛型返回值类型，你可以任意指定。

runnable接口实现的没有返回值的并发编程。callable实现的存在返回值的并发编程。（call的返回值String受泛型的影响）使用Future获取返回值。

执行器服务 ExecutorService

java.util.concurrent.ExecutorService 接口表示一个异步执行机制，使我们能够在后台执行任务。因此一个 ExecutorService 很类似于一个线程池。实际上，存在于 java.util.concurrent 包里的 ExecutorService 实现就是一个线程池实现。以下是一个简单的 ExecutorService 例子：

ExecutorService executorService = Executors.newFixedThreadPool();
executorService.execute(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});
executorService.shutdown();

首先使用 newFixedThreadPool() 工厂方法创建一个 ExecutorService。这里创建了一个十个线程执行任务的线程池。然后，将一个 Runnable 接口的匿名实现类传递给 execute() 方法。这将导致 ExecutorService 中的某个线程执行该 Runnable。

任务委派

下图说明了一个线程是如何将一个任务委托给一个 ExecutorService 去异步执行的：

一个线程将一个任务委派给一个 ExecutorService 去异步执行。一旦该线程将任务委派给 ExecutorService，该线程将继续它自己的执行，独立于该任务的执行。

ExecutorService 实现

既然 ExecutorService 是个接口，如果你想用它的话就得去使用它的实现类之一。java.util.concurrent 包提供了 ExecutorService 接口的以下实现类：

ThreadPoolExecutor（可以使用 Executors 工厂类来创建）
ScheduledThreadPoolExecutor

创建一个 ExecutorService

ExecutorService 的创建依赖于你使用的具体实现。但是你也可以使用 Executors 工厂类来创建 ExecutorService 实例。

以下是几个创建 ExecutorService 实例的例子：

// 创建单一线程池
ExecutorService executorService1 = Executors.newSingleThreadExecutor();
// 创建固定数目线程池
ExecutorService executorService2 = Executors.newFixedThreadPool();
// 创建执行计划线程池
ExecutorService executorService3 = Executors.newScheduledThreadPool();
//创建缓存线程池
ExecutorService executorService4 = Executors.newCachedThreadPool（）

ExecutorService 使用

有几种不同的方式来将任务委托给 ExecutorService 去执行：

execute(Runnable)
submit(Runnable)
submit(Callable)
invokeAny(...)
invokeAll(...)

接下来我们挨个看一下这些方法。

execute(Runnable)

execute(Runnable) 方法要求一个 java.lang.Runnable 对象，然后对它进行异步执行。

ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.execute(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});
executorService.shutdown();

没有办法得知被执行的 Runnable 的执行结果。如果有需要的话你得使用一个 Callable。

submit(Runnable)

submit(Runnable) 方法也要求一个 Runnable 实现类，但它返回一个 Future 对象。这个 Future 对象可以用来检查 Runnable 是否已经执行完毕。

Future future = executorService.submit(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});
future.get();  //returns null if the task has finished correctly

submit(Callable)

submit(Callable) 方法类似于 submit(Runnable) 方法，除了它所要求的参数类型之外。Callable 实例除了它的 call() 方法能够返回一个结果之外和一个 Runnable 很相像。Runnable.run() 不能够返回一个结果。Callable 的结果可以通过 submit(Callable) 方法返回的 Future 对象进行获取。

Future future = executorService.submit(new Callable(){
    public Object call() throws Exception {
        System.out.println("Asynchronous Callable");
        return "Callable Result";
    }
});
System.out.println("future.get() = " + future.get());

以上代码输出：

Asynchronous Callable
future.get() = Callable Result

invokeAny()

invokeAny() 方法要求一系列的 Callable 或者其子接口的实例对象。调用这个方法并不会返回一个 Future，但它返回其中一个 Callable 对象的结果。无法保证返回的是哪个 Callable 的结果 - 只能表明其中一个已执行结束。如果其中一个任务执行结束(或者抛了一个异常)，其他 Callable 将被取消：

ExecutorService executorService = Executors.newSingleThreadExecutor();
Set> callables = new HashSet>();
callables.add(new Callable() {
    public String call() throws Exception {
        return "Task 1";
    }
});
callables.add(new Callable() {
    public String call() throws Exception {
        return "Task 2";
    }
});
callables.add(new Callable() {
    public String call() throws Exception {
        return "Task 3";
    }
});
String result = executorService.invokeAny(callables);
System.out.println("result = " + result);
executorService.shutdown();

上述代码将会打印出给定 Callable 集合中的一个的执行结果。我自己试着执行了它几次，结果始终在变。有时是 "Task 1"，有时是 "Task 2" 等等。

invokeAll()

invokeAll() 方法将调用你在集合中传给 ExecutorService 的所有 Callable 对象。invokeAll() 返回一系列的 Future 对象，通过它们你可以获取每个 Callable 的执行结果。记住，一个任务可能会由于一个异常而结束，因此它可能没有 "成功"。无法通过一个 Future 对象来告知我们是两种结束中的哪一种。

ExecutorService executorService = Executors.newSingleThreadExecutor();
Set> callables = new HashSet>();
callables.add(new Callable() {
    public String call() throws Exception {
        return "Task 1";
    }
});
callables.add(new Callable() {
    public String call() throws Exception {
        return "Task 2";
    }
});
callables.add(new Callable() {
    public String call() throws Exception {
        return "Task 3";
    }
});
List> futures = executorService.invokeAll(callables);
for(Future future : futures){
    System.out.println("future.get = " + future.get());
}
executorService.shutdown();

ExecutorService 关闭

使用完 ExecutorService 之后你应该将其关闭，以使其中的线程不再运行。

executorService.shutdown();

比如，如果你的应用是通过一个 main() 方法启动的，之后 main 方法退出了你的应用，如果你的应用有一个活动的 ExexutorService 它将还会保持运行。ExecutorService 里的活动线程阻止了 JVM 的关闭。要终止 ExecutorService 里的线程你需要调用 ExecutorService 的 shutdown() 方法。

ExecutorService 并不会立即关闭，但它将不再接受新的任务，而且一旦所有线程都完成了当前任务的时候，ExecutorService 将会关闭。在 shutdown() 被调用之前所有提交给 ExecutorService 的任务都被执行。

如果你想要立即关闭 ExecutorService，你可以调用 shutdownNow() 方法。这样会立即尝试停止所有执行中的任务，并忽略掉那些已提交但尚未开始处理的任务。无法担保执行任务的正确执行。可能它们被停止了，也可能已经执行结束。

2. 线程池执行者 ThreadPoolExecutor

java.util.concurrent.ThreadPoolExecutor 是 ExecutorService 接口的一个实现。ThreadPoolExecutor 使用其内部池中的线程执行给定任务(Callable 或者 Runnable)。

ThreadPoolExecutor 包含的线程池能够包含不同数量的线程。池中线程的数量由以下变量决定：

corePoolSize
maximumPoolSize

当一个任务委托给线程池时，如果池中线程数量低于 corePoolSize，一个新的线程将被创建，即使池中可能尚有空闲线程。如果内部任务队列已满，而且有至少 corePoolSize 正在运行，但是运行线程的数量低于 maximumPoolSize，一个新的线程将被创建去执行该任务。

一个 ThreadPoolExecutor。

创建一个 ThreadPoolExecutor

ThreadPoolExecutor 有若干个可用构造子。比如：

int  corePoolSize  =    ;
int  maxPoolSize   =   ;
long keepAliveTime = ;
ExecutorService threadPoolExecutor =
        new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                keepAliveTime,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue()
                );

但是，除非你确实需要显式为 ThreadPoolExecutor 定义所有参数，使用java.util.concurrent.Executors 类中的工厂方法之一会更加方便，正如ExecutorService小节所述。

3. 定时执行者服务 ScheduledExecutorService

java.util.concurrent.ScheduledExecutorService 是一个 ExecutorService，它能够将任务延后执行，或者间隔固定时间多次执行。任务由一个工作者线程异步执行，而不是由提交任务给 ScheduledExecutorService 的那个线程执行。

ScheduledExecutorService 例子

以下是一个简单的 ScheduledExecutorService 示例：

ScheduledExecutorService scheduledExecutorService =
        Executors.newScheduledThreadPool();
ScheduledFuture scheduledFuture =
    scheduledExecutorService.schedule(new Callable() {
        public Object call() throws Exception {
            System.out.println("Executed!");
            return "Called!";
        }
    },
    ,
    TimeUnit.SECONDS);

首先一个内置 5 个线程的 ScheduledExecutorService 被创建。之后一个 Callable 接口的匿名类示例被创建然后传递给 schedule() 方法。后边的俩参数定义了 Callable 将在 5 秒钟之后被执行。

ScheduledExecutorService 实现

既然 ScheduledExecutorService 是一个接口，你要用它的话就得使用 java.util.concurrent 包里对它的某个实现类。ScheduledExecutorService 具有以下实现类：ScheduledThreadPoolExecutor

创建一个 ScheduledExecutorService
如何创建一个 ScheduledExecutorService 取决于你采用的它的实现类。但是你也可以使用 Executors 工厂类来创建一个 ScheduledExecutorService 实例。比如：

ScheduledExecutorService scheduledExecutorService =
  Executors.newScheduledThreadPool();

ScheduledExecutorService 使用

一旦你创建了一个 ScheduledExecutorService，你可以通过调用它的以下方法：

schedule (Callable task, long delay, TimeUnit timeunit)
schedule (Runnable task, long delay, TimeUnit timeunit)
scheduleAtFixedRate (Runnable, long initialDelay, long period, TimeUnit timeunit)
scheduleWithFixedDelay (Runnable, long initialDelay, long period, TimeUnit timeunit)

下面我们就简单看一下这些方法。

schedule (Callable task, long delay, TimeUnit timeunit)

这个方法计划指定的 Callable 在给定的延迟之后执行。这个方法返回一个 ScheduledFuture，通过它你可以在它被执行之前对它进行取消，或者在它执行之后获取结果。

ScheduledExecutorService scheduledExecutorService =
        Executors.newScheduledThreadPool();
ScheduledFuture scheduledFuture =
    scheduledExecutorService.schedule(new Callable() {
        public Object call() throws Exception {
            System.out.println("Executed!");
            return "Called!";
        }
    },
    ,
    TimeUnit.SECONDS);
System.out.println("result = " + scheduledFuture.get());
scheduledExecutorService.shutdown();

示例输出结果：

Executed!
result = Called!

schedule (Runnable task, long delay, TimeUnit timeunit)

除了 Runnable 无法返回一个结果之外，这一方法工作起来就像以一个 Callable 作为一个参数的那个版本的方法一样，因此 ScheduledFuture.get() 在任务执行结束之后返回 null。

scheduleAtFixedRate (Runnable, long initialDelay, long period, TimeUnit timeunit)

这一方法规划一个任务将被定期执行。该任务将会在首个 initialDelay 之后得到执行，然后每个 period 时间之后重复执行。如果给定任务的执行抛出了异常，该任务将不再执行。如果没有任何异常的话，这个任务将会持续循环执行到 ScheduledExecutorService 被关闭。如果一个任务占用了比计划的时间间隔更长的时候，下一次执行将在当前执行结束执行才开始。计划任务在同一时间不会有多个线程同时执行。

scheduleWithFixedDelay (Runnable, long initialDelay, long period, TimeUnit timeunit)

除了 period 有不同的解释之外这个方法和 scheduleAtFixedRate() 非常像。

scheduleAtFixedRate() 方法中，period 被解释为前一个执行的开始和下一个执行的开始之间的间隔时间。而在本方法中，period 则被解释为前一个执行的结束和下一个执行的结束之间的间隔。因此这个延迟是执行结束之间的间隔，而不是执行开始之间的间隔。

ScheduledExecutorService 关闭

正如 ExecutorService，在你使用结束之后你需要把 ScheduledExecutorService 关闭掉。否则他将导致 JVM 继续运行，即使所有其他线程已经全被关闭。你可以使用从 ExecutorService 接口继承来的 shutdown() 或 shutdownNow() 方法将 ScheduledExecutorService 关闭。参见 ExecutorService 关闭部分以获取更多信息。

4. 使用 ForkJoinPool 进行分叉和合并

ForkJoinPool 在 Java 7 中被引入。它和ExecutorService很相似，除了一点不同。ForkJoinPool 让我们可以很方便地把任务分裂成几个更小的任务，这些分裂出来的任务也将会提交给 ForkJoinPool。任务可以继续分割成更小的子任务，只要它还能分割。可能听起来有些抽象，因此本节中我们将会解释 ForkJoinPool 是如何工作的，还有任务分割是如何进行的。

分叉和合并解释

在我们开始看 ForkJoinPool 之前我们先来简要解释一下分叉和合并的原理。分叉和合并原理包含两个递归进行的步骤。两个步骤分别是分叉步骤和合并步骤。

分叉

一个使用了分叉和合并原理的任务可以将自己分叉(分割)为更小的子任务，这些子任务可以被并发执行。如下图所示

通过把自己分割成多个子任务，每个子任务可以由不同的 CPU 并行执行，或者被同一个 CPU 上的不同线程执行。只有当给的任务过大，把它分割成几个子任务才有意义。把任务分割成子任务有一定开销，因此对于小型任务，这个分割的消耗可能比每个子任务并发执行的消耗还要大。

什么时候把一个任务分割成子任务是有意义的，这个界限也称作一个阀值。这要看每个任务对有意义阀值的决定。很大程度上取决于它要做的工作的种类。

合并

当一个任务将自己分割成若干子任务之后，该任务将进入等待所有子任务的结束之中。一旦子任务执行结束，该任务可以把所有结果合并到同一个结果。图示如下：

当然，并非所有类型的任务都会返回一个结果。如果这个任务并不返回一个结果，它只需等待所有子任务执行完毕。也就不需要结果的合并啦。

ForkJoinPool

ForkJoinPool 是一个特殊的线程池，它的设计是为了更好的配合分叉-和-合并任务分割的工作。ForkJoinPool 也在 java.util.concurrent 包中，其完整类名为 java.util.concurrent.ForkJoinPool。

创建一个 ForkJoinPool

你可以通过其构造子创建一个 ForkJoinPool。作为传递给 ForkJoinPool 构造子的一个参数，你可以定义你期望的并行级别。并行级别表示你想要传递给 ForkJoinPool 的任务所需的线程或 CPU 数量。以下是一个 ForkJoinPool 示例：

ForkJoinPool forkJoinPool = new ForkJoinPool();

这个示例创建了一个并行级别为 4 的 ForkJoinPool。

提交任务到 ForkJoinPool

就像提交任务到 ExecutorService 那样，把任务提交到 ForkJoinPool。你可以提交两种类型的任务。一种是没有任何返回值的(一个 "行动")，另一种是有返回值的(一个"任务")。这两种类型分别由 RecursiveAction 和 RecursiveTask 表示。接下来介绍如何使用这两种类型的任务，以及如何对它们进行提交。

RecursiveAction

RecursiveAction 是一种没有任何返回值的任务。它只是做一些工作，比如写数据到磁盘，然后就退出了。一个 RecursiveAction 可以把自己的工作分割成更小的几块，这样它们可以由独立的线程或者 CPU 执行。你可以通过继承来实现一个 RecursiveAction。示例如下：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.RecursiveAction;
public class MyRecursiveAction extends RecursiveAction {
    private long workLoad = ;
    public MyRecursiveAction(long workLoad) {
        this.workLoad = workLoad;
    }
    @Override
    protected void compute() {
        //if work is above threshold, break tasks up into smaller tasks
        if(this.workLoad > ) {
            System.out.println("Splitting workLoad : " + this.workLoad);
            List subtasks =
                new ArrayList();
            subtasks.addAll(createSubtasks());
            for(RecursiveAction subtask : subtasks){
                subtask.fork();
            }
        } else {
            System.out.println("Doing workLoad myself: " + this.workLoad);
        }
    }
    private List createSubtasks() {
        List subtasks =
            new ArrayList();
        MyRecursiveAction subtask1 = new MyRecursiveAction(this.workLoad / );
        MyRecursiveAction subtask2 = new MyRecursiveAction(this.workLoad / );
        subtasks.add(subtask1);
        subtasks.add(subtask2);
        return subtasks;
    }
}

例子很简单。MyRecursiveAction 将一个虚构的 workLoad 作为参数传给自己的构造子。如果 workLoad 高于一个特定阀值，该工作将被分割为几个子工作，子工作继续分割。如果 workLoad 低于特定阀值，该工作将由 MyRecursiveAction 自己执行。你可以这样规划一个 MyRecursiveAction 的执行：

MyRecursiveAction myRecursiveAction = new MyRecursiveAction();
forkJoinPool.invoke(myRecursiveAction);

RecursiveTask

RecursiveTask 是一种会返回结果的任务。它可以将自己的工作分割为若干更小任务，并将这些子任务的执行结果合并到一个集体结果。可以有几个水平的分割和合并。以下是一个 RecursiveTask 示例：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.RecursiveTask;
public class MyRecursiveTask extends RecursiveTask {
    private long workLoad = ;
    public MyRecursiveTask(long workLoad) {
        this.workLoad = workLoad;
    }
    protected Long compute() {
        //if work is above threshold, break tasks up into smaller tasks
        if(this.workLoad > ) {
            System.out.println("Splitting workLoad : " + this.workLoad);
            List subtasks =
                new ArrayList();
            subtasks.addAll(createSubtasks());
            for(MyRecursiveTask subtask : subtasks){
                subtask.fork();
            }
            long result = ;
            for(MyRecursiveTask subtask : subtasks) {
                result += subtask.join();
            }
            return result;
        } else {
            System.out.println("Doing workLoad myself: " + this.workLoad);
            return workLoad * ;
        }
    }
    private List createSubtasks() {
        List subtasks =
        new ArrayList();
        MyRecursiveTask subtask1 = new MyRecursiveTask(this.workLoad / );
        MyRecursiveTask subtask2 = new MyRecursiveTask(this.workLoad / );
        subtasks.add(subtask1);
        subtasks.add(subtask2);
        return subtasks;
    }
}

除了有一个结果返回之外，这个示例和 RecursiveAction 的例子很像。MyRecursiveTask 类继承自 RecursiveTask，这也就意味着它将返回一个 Long 类型的结果。

MyRecursiveTask 示例也会将工作分割为子任务，并通过 fork() 方法对这些子任务计划执行。

此外，本示例还通过调用每个子任务的 join() 方法收集它们返回的结果。子任务的结果随后被合并到一个更大的结果，并最终将其返回。对于不同级别的递归，这种子任务的结果合并可能会发生递归。

你可以这样规划一个 RecursiveTask：

MyRecursiveTask myRecursiveTask = new MyRecursiveTask();
long mergedResult = forkJoinPool.invoke(myRecursiveTask);
System.out.println("mergedResult = " + mergedResult);

注意是如何通过 ForkJoinPool.invoke() 方法的调用来获取最终执行结果的。

Java 线程池 ExecutorService

Java 线程池 Java通过Executors提供四种线程池：

Java相关栏目本月热门文章

Java 线程池
Java通过Executors提供四种线程池：