java.util.concurrent - Java 并发工具

作者: liycode | 来源:发表于2017-03-12 17:35 被阅读72次

翻译自http://tutorials.jenkov.com/java-util-concurrent/index.html

java.util.concurrent - Java 并发工具包一

一、ExecutorService（执行器服务）

java.util.concurrent.ExecutorService接口代表了一种异步执行机制，它可以在后台执行任务。一个ExecutorService接口与线程池是非常相似的。实际上，线程池就是实现ExecutorService的一个实现。

1.1 ExecutorService示例

ExecutorService executorService = Executors.newFixedThreadPool(10);

executorService.execute(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});

executorService.shutdown();

ExecutorService通过Executors.newFixedThreadPool()工厂方法被创建的。这个示例中创建了一个有10个线程的线程池来执行任务。

一个匿名的Runnable接口实现被传递给execute()方法。这个Runnable将会被ExecutorService中的一个线程来执行。

1.2 任务委托

下面是一个图示，一个线程委托一个任务到ExecutorService：

[图片上传失败...(image-b953fd-1514260695995)]

线程1委托任务后将会继续它自己的操作，与任务已经没有关系。

1.3 ExecutorService实现

ExecutorService既然是一个接口，必须实现后才能使用。java.util.concurrent包里有如下实现：

ThreadPoolExecutor
ScheduledThreadPoolExecutor

1.4 创建ExecutorService

创建ExecutorService取决于你使用的实现类。可以使用Executors工厂类来创建ExecutorService，下面是一些例子：

ExecutorService executorService1 = Executors.newSingleThreadExecutor();

ExecutorService executorService2 = Executors.newFixedThreadPool(10);

ExecutorService executorService3 = Executors.newScheduledThreadPool(10);

根据方法的名字很容易明白其含义。

1.5 ExecutorService使用方法

下面是一些委托任务的方式：

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

execute(Runnable)

这个方法需要一个java.lang.Runnable对象，与上面的例子一样：

ExecutorService executorService = Executors.newSingleThreadExecutor();

executorService.execute(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});

executorService.shutdown();

这种方法获得不到Runnable的执行结果，如果需要执行结果则需使用submit(Callable)。

submit(Runnable)

submit(Runnable)方法也接收一个Runnable实现，但是返回一个Future对象。这个Future对象可以用来检查任务是否执行完毕。

下面是一个例子：

Future future = executorService.submit(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});

future.get();  // 如果任务正确执行完会返回 null

注意，它仍然不能获得返回的结果，只能判断任务是否执行完。

submit(Callable)

它与submit(Runnable)类似，Callable接口也与Runnable类似，只不过Callable的call()方法可以返回一个结果，Runnable.run()却不行。

返回的结果仍然通过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对象的集合，或者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";
    }
});

String result = executorService.invokeAny(callables);

System.out.println("result = " + result);

executorService.shutdown();

这个例子将会打印出某个Callable返回的结果。如果重复运行本例子得到的结果可能会变化，有时候是Task 1有时候是Task 2。

invokeAll()

invokeAll()会调用Callable集合里的所有对象，并且返回一个List<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();

1.6 ExecutorService关闭

当你使用完ExecutorService后应该使用shutdown()方法关闭它，这样线程就不会继续运行了。ExecutorService并不会立即关闭，但是它也不会再接收新任务，当所有的线程完成了它们的任务后ExecutorService就会关闭。

如果你想立即结束ExecutorService应该调用shutdownNow()方法，但是这个并没有保障，也许它们会停止也许会执行完。

二、ThreadPoolExecutor（线程池执行器）

java.util.concurrent.ThreadPoolExecutor是ExecutorService接口的实现。ThreadPoolExecutor使用内部线程池中的线程执行传递来的任务。

线程池可以包含不同数量的线程。线程的数量被下面这些参数决定：

corePoolSize
maximumPoolSize

如果当一个任务被委托给线程池时线程池中线程的数量小于corePoolSize，一个新的线程会被创建，就算池中有空闲的线程。

如果内部的任务队列已经满了，并且corePoolSize或者更多的线程（小于maximumPoolSize）在运行，那么就会创建一个新线程去执行任务。

ThreadPoolExecutor原理

2.1 创建一个ThreadPoolExecutor

ThreadPoolExecutor有多个构造器，例如：

int  corePoolSize  =    5;
int  maxPoolSize   =   10;
long keepAliveTime = 5000;

ExecutorService threadPoolExecutor =
        new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                keepAliveTime,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>()
                );

注意，除非你要明确的指定所有参数，通常使用java.util.concurrent.Executors中的工厂方法是更好的方式。

例如：

ExecutorService executorService = Executors.newFixedThreadPool(10);

三、ScheduledExecutorService

java.util.concurrent.ScheduledExecutorService接口继承自ExecutorService，可以安排任务在一定延迟后执行，或者间隔固定时间重复执。任务被线程池中的一个线程执行，并不是将任务传递给ScheduledExecutorService的那个线程执行。

下面是一个例子：

// 线程池中有5个线程
ScheduledExecutorService scheduledExecutorService =
        Executors.newScheduledThreadPool(5);

// 5秒后执行Callable方法
ScheduledFuture scheduledFuture =
    scheduledExecutorService.schedule(new Callable() {
        public Object call() throws Exception {
            System.out.println("Executed!");
            return "Called!";
        }
    },
    5,
    TimeUnit.SECONDS);

ScheduledExecutorService对应的实现类是ScheduledThreadPoolExecutor。

可以使用工厂类创建：

ScheduledExecutorService scheduledExecutorService =
    Executors.newScheduledThreadPool(5);

用例

一旦你创建了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);

// scheduledFuture.get()将会等待5秒后打印结果
System.out.println("result = " + scheduledFuture.get());

scheduledExecutorService.shutdown();

schedule (Runnable task, long delay, TimeUnit timeunit)

与上一个方法类似，但是一个Runnable不能返回一个值，当方法结束时ScheduledFuture.get()将会返回null。

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

使某个任务被周期性执行。这个任务起初在延迟initialDelay时间后开始执行，然后以period时间长度开始循环。

任何异常都会终止循环，如果没有异常任务将会一直执行，直到ScheduledExecutorService被关闭。如果任务执行所需时间大于period，下一次任务的执行将会在当前任务执行结束后。任务同一时间不会被多个线程执行。

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

这个方法的工作方式与scheduleAtFixedRate()非常类似，不同的在于对period的解释上。

scheduleAtFixedRate()将period解释为每个方法开始执行的时间间隔。例如：A开始执行后，过5秒将再次执行A。

但是在本方法中将period解释为每个方法执行结束后的时间间隔。例如：A执行结束后，过5秒将再次执行A。

ScheduledExecutorService关闭

就像ExecutorService一样，当你不使用ScheduledExecutorService时需要将其关闭。如果不这样做它将会一直运行，就算所有其他线程都被关闭。

四、通过ForkJoinPool使用Java Fork和Join

ForkJoinPool在Java 7中被添加进来。ForkJoinPool与ExecutorService类似，ForkJoinPool使任务更容易被分割成小的任务，然后提交到ForkJoinPool中。小任务可以继续分割下去。

4.1 Fork和Join

Fork

一个任务使用fork和join原理，可以fork (分叉)它自己成为更小的子任务，这些子任务可以被并发执行。请参考下图：

image

如果某个任务的工作很巨大，它将自己分裂成数个子任务才有意义。将任务分裂成子任务是有开销的，所以对于小的任务进行分裂，这种开销可能比异步执行的子任务的开销还大。

将任务分裂成子任务是有意义的限制也被称为门槛。它很依赖于工作的类型。

Join

当一个任务将其自身分叉为多个子任务，这个任务将会等待，直到所有子任务执行完成。

一旦子任务执行结束，父任务将会合并所有的结果到一个结果。如下图所示：

image

当然不是所有类型的任务都会返回一个结果。如果一个任务没有返回值那么它只要等待所有子任务执行完毕即可。没有结果合并会发生。

4.2 ForkJoinPool

java.util.concurrent.ForkJoinPool是一个特殊的线程池，它被设计用来分叉合并任务。

创建一个ForkJoinPool

使用ForkJoinPool的构造器来创建它。构造器的参数用来声明并发级别。并发级别指定了你想使用多少个线程来执行任务。

ForkJoinPool forkJoinPool = new ForkJoinPool(4);

提交任务到ForkJoinPool

与提交任务到ExecutorService类似。你可以提交2种类型的任务。一种是无返回值的任务（action）另一种是有返回值的任务（task），相应的类为：RecursiveAction、RecursiveTask。

RecursiveAction

RecursiveAction是一个不返回任何值的任务。它只是做一些工作，例如写数据到硬盘然后退出。

未完待续...

更多文章请访问我的网站liycode.com

java.util.concurrent - Java 并发工具

一、ExecutorService（执行器服务）

1.1 ExecutorService示例

1.2 任务委托

1.3 ExecutorService实现

1.4 创建ExecutorService

1.5 ExecutorService使用方法

execute(Runnable)

submit(Runnable)

submit(Callable)

invokeAny()

invokeAll()

1.6 ExecutorService关闭

二、ThreadPoolExecutor（线程池执行器）

2.1 创建一个ThreadPoolExecutor

三、ScheduledExecutorService

四、通过ForkJoinPool使用Java Fork和Join

4.1 Fork和Join

4.2 ForkJoinPool

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Java并发

Java