• 极客专栏正式上线!欢迎访问 https://www.jikewenku.com/topic.html
  • 极客专栏正式上线!欢迎访问 https://www.jikewenku.com/topic.html

并发编程 :Concurrent 用户指南 ( 中 )

技术杂谈 勤劳的小蚂蚁 3个月前 (02-06) 64次浏览 已收录 0个评论 扫描二维码

16. 执行器服务 ExecutorService

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

ExecutorService 例子

以下是一个简单的 ExecutorService 例子:

ExecutorService executorService = Executors.newFixedThreadPool(10);  
 
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
  • ScheduledThreadPoolExecutor

创建一个 ExecutorService

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

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

ExecutorService executorService1 = Executors.newSingleThreadExecutor();  
 
ExecutorService executorService2 = Executors.newFixedThreadPool(10);  
 
ExecutorService executorService3 = Executors.newScheduledThreadPool(10);

ExecutorService 使用

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

  • execute(Runnable)
  • submit(Runnable)
  • submit(Callable)
  • invokeAny(…)
  • invokeAll(…)

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

execute(Runnable)

execute(Runnable) 方法要求一个 java.lang.Runnable 对象,然后对它进行异步执行。以下是使用 ExecutorService 执行一个 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 是否已经执行完毕。以下是 ExecutorService submit() 示例:

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 对象进行获取。

以下是一个 ExecutorService Callable 示例:

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<Callable<String>> callables = new HashSet<Callable<String>>();  
 
callables.add(new Callable<String>() {  
    public String call() throws Exception {  
        return “Task 1”;  
    }  
});  
callables.add(new Callable<String>() {  
    public String call() throws Exception {  
        return “Task 2”;  
    }  
});  
callables.add(new Callable<String>() {  
    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<Callable<String>> callables = new HashSet<Callable<String>>();  
 
callables.add(new Callable<String>() {  
    public String call() throws Exception {  
        return “Task 1”;  
    }  
});  
callables.add(new Callable<String>() {  
    public String call() throws Exception {  
        return “Task 2”;  
    }  
});  
callables.add(new Callable<String>() {  
    public String call() throws Exception {  
        return “Task 3”;  
    }  
});  
 
List<Future<String>> futures = executorService.invokeAll(callables);  
 
for(Future<String> future : futures){  
    System.out.println(“future.get = ” + future.get());  
}  
 
executorService.shutdown();

ExecutorService 关闭

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

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

要终止 ExecutorService 里的线程你需要调用 ExecutorService 的 shutdown() 方法。ExecutorService 并不会立即关闭,但它将不再接受新的任务,而且一旦所有线程都完成了当前任务的时候,ExecutorService 将会关闭。在 shutdown() 被调用之前所有提交给 ExecutorService 的任务都被执行。如果你想要立即关闭 ExecutorService,你可以调用 shutdownNow() 方法。这样会立即尝试停止所有执行中的任务,并忽略掉那些已提交但尚未开始处理的任务。无法担保执行任务的正确执行。可能它们被停止了,也可能已经执行结束。

17. 线程池执行者 ThreadPoolExecutor

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

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

  • corePoolSize
  • maximumPoolSize

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

ThreadPoolExecutor 图解:


一个 ThreadPoolExecutor

创建一个 ThreadPoolExecutor

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

int  corePoolSize  =    5;  
int  maxPoolSize   =   10;  
long keepAliveTime = 5000;  
 
ExecutorService threadPoolExecutor =  
        new ThreadPoolExecutor(  
                corePoolSize,  
                maxPoolSize,  
                keepAliveTime,  
                TimeUnit.MILLISECONDS,  
                new LinkedBlockingQueue<Runnable>()  
                );

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

18. 定时执行者服务 ScheduledExecutorService

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

ScheduledExecutorService 例子

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

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

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

ScheduledExecutorService 实现

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

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

ScheduledExecutorService scheduledExecutorService =  
 
        Executors.newScheduledThreadPool(5);

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(5);  
 
ScheduledFuture scheduledFuture =  
    scheduledExecutorService.schedule(new Callable() {  
        public Object call() throws Exception {  
            System.out.println(“Executed!”);  
            return “Called!”;  
        }  
    },  
    5,  
    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 关闭部分以获取更多信息。

19. 使用 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);

这个示例创建了一个并行级别为 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 = 0;  
 
    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 > 16) {  
            System.out.println(“Splitting workLoad : ” + this.workLoad);  
 
            List<MyRecursiveAction> subtasks =  
                new ArrayList<MyRecursiveAction>();  
 
            subtasks.addAll(createSubtasks());  
 
            for(RecursiveAction subtask : subtasks){  
                subtask.fork();  
            }  
 
        } else {  
            System.out.println(“Doing workLoad myself: ” + this.workLoad);  
        }  
    }  
 
    private List<MyRecursiveAction> createSubtasks() {  
        List<MyRecursiveAction> subtasks =  
            new ArrayList<MyRecursiveAction>();  
 
        MyRecursiveAction subtask1 = new MyRecursiveAction(this.workLoad / 2);  
        MyRecursiveAction subtask2 = new MyRecursiveAction(this.workLoad / 2);  
 
        subtasks.add(subtask1);  
        subtasks.add(subtask2);  
 
        return subtasks;  
    }  
 
}

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

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

RecursiveTask

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

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

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

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

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

你可以这样规划一个 RecursiveTask:

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

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

ForkJoinPool 评论

貌似并非每个人都对 Java 7 里的 ForkJoinPool 满意:《一个 Java 分叉-合并 带来的灾祸》。


在你计划在自己的项目里使用 ForkJoinPool 之前最好读一下该篇文章。


丨极客文库, 版权所有丨如未注明 , 均为原创丨
本网站采用知识共享署名-非商业性使用-相同方式共享 3.0 中国大陆许可协议进行授权
转载请注明原文链接:并发编程 :Concurrent 用户指南 ( 中 )
喜欢 (0)
[247507792@qq.com]
分享 (0)
勤劳的小蚂蚁
关于作者:
温馨提示:本文来源于网络,转载文章皆标明了出处,如果您发现侵权文章,请及时向站长反馈删除。

您必须 登录 才能发表评论!

  • 精品技术教程
  • 编程资源分享
  • 问答交流社区
  • 极客文库知识库

客服QQ


QQ:2248886839


工作时间:09:00-23:00