Java并发的四种风味：Thread、Executor、ForkJoin和Actor

这篇文章讨论了Java应用中并行处理的多种方法。从自己管理Java线程，到各种更好几的解决方法，Executor服务、ForkJoin 框架以及计算中的Actor模型。

Java并发编程的4种风格：Threads，Executors，ForkJoin和Actors

我们生活在一个事情并行发生的世界。自然地，我们编写的程序也反映了这个特点，它们可以并发的执行。当然除了Python代码（译者注：链接里面讲述了Python的全局解释器锁，解释了原因），不过你仍然可以使用Jython在JVM上运行你的程序，来利用多处理器电脑的强大能力。

然而，并发程序的复杂程度远远超出了人类大脑的处理能力。相比较而言，我们简直弱爆了：我们生来就不是为了思考多线程程序、评估并发访问有限资源以及预测哪里会发生错误或者瓶颈。

面对这些困难，人类已经总结了不少并发计算的解决方案和模型。这些模型强调问题的不同部分，当我们实现并行计算时，可以根据问题做出不同的选择。

在这篇文章中，我将会用对同一个问题，用不同的代码来实现并发的解决方案；然后讨论这些方案有哪些好的地方，有哪些缺陷，可能会有什么样的陷阱在等着你。

我们将介绍下面几种并发处理和异步代码的方式：

• 裸线程

• Executors和Services

• ForkJoin框架和并行流

• Actor模型

为了更加有趣一些，我没有仅仅通过一些代码来说明这些方法，而是使用了一个共同的任务，因此每一节中的代码差不多都是等价的。另外，这些代码仅仅是展示用的，初始化的代码并没有写出来，并且它们也不是产品级的软件示例。

对了，最后一件事：在文章最后，有一个小调查，关于你或者你的组织正在使用哪种并发模式。为了你的工程师同胞们，请填一下调查！

任务

任务：实现一个方法，它接收一条消息和一组字符串作为参数，这些字符串与某个搜索引擎的查询页面对应。对每个字符串，这个方法发出一个http请求来查询消息，并返回第一条可用的结果，越快越好。

如果有错误发生，抛出一个异常或者返回空都是可以的。我只是尝试避免为了等待结果而出现无限循环。

简单说明：这次我不会真正深入到多线程如何通讯的细节，或者深入到Java内存模型。如果你迫切地想了解这些，你可以看我前面的文章利用JCStress测试并发。

那么，让我们从最直接、最核心的方式来在JVM上实现并发：手动管理裸线程。

方法1：使用“原汁原味”的裸线程

解放你的代码，回归自然，使用裸线程！线程是并发最基本的单元。Java线程本质上被映射到操作系统线程，并且每个线程对象对应着一个计算机底层线程。

自然地，JVM管理着线程的生存期，而且只要你不需要线程间通讯，你也不需要关注线程调度。

每个线程有自己的栈空间，它占用了JVM进程空间的指定一部分。

线程的接口相当简明，你只需要提供一个Runnable，调用.start()开始计算。没有现成的API来结束线程，你需要自己来实现，通过类似boolean类型的标记来通讯。

在下面的例子中，我们对每个被查询的搜索引擎，创建了一个线程。查询的结果被设置到AtomicReference，它不需要锁或者其他机制来保证只出现一次写操作。开始吧！


1

2

3

4

5

6

7

8

9

10

11

private static String getFirstResult(String question, List<String> engines) {

 AtomicReference<String> result = new AtomicReference<>();

 for(String base: engines) {

   String url = base + question;

   new Thread(() -> {

     result.compareAndSet(null, WS.url(url).get());

   }).start();

 }

 while(result.get() == null); // wait for some result to appear

 return result.get();

}

使用裸线程的主要优点是，你很接近并发计算的操作系统/硬件模型，并且这个模型非常简单。多个线程运行，通过共享内存通讯，就是这样。

自己管理线程的最大劣势是，你很容易过分的关注线程的数量。线程是很昂贵的对象，创建它们需要耗费大量的内存和时间。这是一个矛盾，线程太少，你不能获得良好的并发性；线程太多，将很可能导致内存问题，调度也变得更复杂。

然而，如果你需要一个快速和简单的解决方案，你绝对可以使用这个方法，不要犹豫。

方法2：认真对待Executor和CompletionService

另一个选择是使用API来管理一组线程。幸运的是，JVM为我们提供了这样的功能，就是Executor接口。Executor接口的定义非常简单：


1

2

3

4

5

public interface Executor {

void execute(Runnable command);

}

它隐藏了如何处理Runnable的细节。它仅仅说，“开发者！你只是一袋肉，给我任务，我会处理它！”

更酷的是，Executors类提供了一组方法，能够创建拥有完善配置的线程池和executor。我们将使用newFixedThreadPool()，它创建预定义数量的线程，并不允许线程数量超过这个预定义值。这意味着，如果所有的线程都被使用的话，提交的命令将会被放到一个队列中等待；当然这是由executor来管理的。

在它的上层，有ExecutorService管理executor的生命周期，以及CompletionService会抽象掉更多细节，作为已完成任务的队列。得益于此，我们不必担心只会得到第一个结果。

下面service.take()的一次调用将会只返回一个结果。


1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

private static String getFirstResultExecutors(String question, List<String> engines) {

 ExecutorCompletionService<String> service = new ExecutorCompletionService<String>(Executors.newFixedThreadPool(4));

 for(String base: engines) {

   String url = base + question;

   service.submit(() -> {

     return WS.url(url).get();

   });

 }

   try {

     return service.take().get();

   }

   catch(InterruptedException | ExecutionException e) {

     return null;

   }

}

如果你需要精确的控制程序产生的线程数量，以及它们的精确行为，那么executor和executor服务将是正确的选择。例如，需要仔细考虑的一个重要问题是，当所有线程都在忙于做其他事情时，需要什么样的策略？增加线程数量或者不做数量限制？把任务放入到队列等待？如果队列也满了呢？无限制的增加队列大小？

感谢JDK，已经有很多配置项回答了这些问题，并且有着直观的名字，例如上面的Executors.newFixedThreadPool(4)。

线程和服务的生命周期也可以通过选项来配置，使资源可以在恰当的时间关闭。唯一的不便之处是，对新手来说，配置选项可以更简单和直观一些。然而，在并发编程方面，你几乎找不到更简单的了。

总之，对于大型系统，我个人认为使用executor最合适。

方法3：通过并行流，使用ForkJoinPool (FJP)

Java 8中加入了并行流，从此我们有了一个并行处理集合的简单方法。它和lambda一起，构成了并发计算的一个强大工具。

如果你打算运用这种方法，那么有几点需要注意。首先，你必须掌握一些函数编程的概念，它实际上更有优势。其次，你很难知道并行流实际上是否使用了超过一个线程，这要由流的具体实现来决定。如果你无法控制流的数据源，你就无法确定它做了什么。

另外，你需要记住，默认情况下是通过ForkJoinPool.commonPool()实现并行的。这个通用池由JVM来管理，并且被JVM进程内的所有线程共享。这简化了配置项，因此你不用担心。


1

2

3

4

5

6

7

8

private static String getFirstResult(String question, List<String> engines) {

 // get element as soon as it is available

 Optional<String> result = engines.stream().parallel().map((base) -> {

   String url = base + question;

   return WS.url(url).get();

 }).findAny();

 return result.get();

}

看上面的例子，我们不关心单独的任务在哪里完成，由谁完成。然而，这也意味着，你的应用程序中可能存在一些停滞的任务，而你却无法不知道。在另一篇关于并行流的文章中，我详细地描述了这个问题。并且有一个变通的解决方案，虽然它并不是世界上最直观的方案。

ForkJoin是一个很好的框架，由比我更聪明的人来编写和预先配置。因此当我需要写一个包含并行处理的小型程序时，它是我的第一选择。

它最大的缺点是，你必须预见到它可能产生的并发症。如果对JVM没有整体上的深入了解，这很难做到。这只能来自于经验。

方法4：雇用一个Actor

Actor模型是对我们本文中所探讨的方法的一个奇怪的补充。JDK中没有actor的实现；因此你必须引用一些实现了actor的库。

简短地说，在actor模型中，你把一切都看做是一个actor。一个actor是一个计算实体，就像上面第一个例子中的线程，它可以从其他actor那里接收消息，因为一切都是actor。

在应答消息时，它可以给其他actor发送消息，或者创建新的actor并与之交互，或者只改变自己的内部状态。

相当简单，但这是一个非常强大的概念。生命周期和消息传递由你的框架来管理，你只需要指定计算单元是什么就可以了。另外，actor模型强调避免全局状态，这会带来很多便利。你可以应用监督策略，例如免费重试，更简单的分布式系统设计，错误容忍度等等。

下面是一个使用Akka Actors的例子。Akka Actors有Java接口，是最流行的JVM Actor库之一。实际上，它也有Scala接口，并且是Scala目前默认的actor库。Scala曾经在内部实现了actor。不少JVM语言都实现了actor，比如Fantom。这些说明了Actor模型已经被广泛接受，并被看做是对语言非常有价值的补充。


1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

static class Message {

 String url;

 Message(String url) {this.url = url;}

}

static class Result {

 String html;

 Result(String html) {this.html = html;}

}

static class UrlFetcher extends UntypedActor {

 @Override

 public void onReceive(Object message) throws Exception {

   if (message instanceof Message) {

     Message work = (Message) message;

     String result = WS.url(work.url).get();

     getSender().tell(new Result(result), getSelf());

   } else {

     unhandled(message);

   }

 }

}

static class Querier extends UntypedActor {

 private String question;

 private List<String> engines;

 private AtomicReference<String> result;

 public Querier(String question, List<String> engines, AtomicReference<String> result) {

   this.question = question;

   this.engines = engines;

   this.result = result;

 }

 @Override public void onReceive(Object message) throws Exception {

   if(message instanceof Result) {

     result.compareAndSet(null, ((Result) message).html);

     getContext().stop(self());

   }

   else {

     for(String base: engines) {

       String url = base + question;

       ActorRef fetcher = this.getContext().actorOf(Props.create(UrlFetcher.class), "fetcher-"+base.hashCode());

       Message m = new Message(url);

       fetcher.tell(m, self());

     }

   }

 }

}

private static String getFirstResultActors(String question, List<String> engines) {

 ActorSystem system = ActorSystem.create("Search");

 AtomicReference<String> result = new AtomicReference<>();

 final ActorRef q = system.actorOf(

   Props.create((UntypedActorFactory) () -> new Querier(question, engines, result)), "master");

 q.tell(new Object(), ActorRef.noSender());

 while(result.get() == null);

 return result.get();

}

Akka actor在内部使用ForkJoin框架来处理工作。这里的代码很冗长。不要担心。大部分代码是消息类Message和Result的定义，然后是两个不同的actor：Querier用来组织所有的搜索引擎，而URLFetcher用来从给定的URL获取结果。这里代码行比较多是因为我不愿意把很多东西写在同一行上。Actor模型的强大之处来自于Props对象的接口，通过接口我们可以为actor定义特定的选择模式，定制的邮箱地址等。结果系统也是可配置的，只包含了很少的活动件。这是一个很好的迹象！

使用Actor模型的一个劣势是，它要求你避免全局状态，因此你必须小心的设计你的应用程序，而这可能会使项目迁移变得很复杂。同时，它也有不少优点，因此学习一些新的范例和使用新的库是完全值得的。

反馈时间：你使用什么？

你最常用的并发方式是什么？你理解它背后的计算模式是什么吗？仅仅使用一个包含Job或者后台任务对象的框架来自动地为你的代码添加异步计算能力？

为了收集更多信息，以找出我是否应该继续更深入地讲解一些不同的并发模式，例如，写一篇关于Akka如何工作，以及它Java接口的优点和缺点，我创建了一个简单的调查。亲爱的读者，请填一下调查表。我非常感谢你的互动！

总结

这篇文章中我们讨论了在Java应用中添加并行的几种不同方法。从我们自己管理Java线程开始，我们逐渐地发现更高级的解决方案，执行不同的executor服务、ForkJoin框架和actor计算模型。

不知道当你面临真实问题时该如何选择？它们都有各自的优缺点，你需要在直观和易用性、配置和增加/减少机器性能等方面做出选择。

时间： 2024-10-28 05:04:56

Java并发的四种风味：Thread、Executor、ForkJoin和Actor的相关文章

[转]Java并发的四种风味：Thread、Executor、ForkJoin和Actor

这篇文章讨论了Java应用中并行处理的多种方法.从自己管理Java线程,到各种更好几的解决方法,Executor服务.ForkJoin 框架以及计算中的Actor模型. Java并发编程的4种风格:Threads,Executors,ForkJoin和Actors 我们生活在一个事情并行发生的世界.自然地,我们编写的程序也反映了这个特点,它们可以并发的执行.当然除了Python代码(译者注:链接里面讲述了Python的全局解释器锁,解释了原因),不过你仍然可以使用Jython在JVM上运行你的程

JAVA中的四种引用以及ReferenceQueue和WeakHashMap的使用示例

简介: 本文主要介绍JAVA中的四种引用: StrongReference(强引用).SoftReferenc(软引用).WeakReferenc(弱引用).PhantomReference(虚引用)的作用.同时我们还将介绍ReferenceQueue和WeakHashMap的功能和使用示例. 欢迎探讨,如有错误敬请指正 如需转载,请注明出处 http://www.cnblogs.com/nullzx/ 1. JAVA中的四种引用 四种引用中,软引用.若引用.虚引用都需要相关类来创建.创建的时候

java中的四种单例实现方式

在java中,单例设计模式是非常常见的设计模式,对单例设计模式的概念,不做过多的介绍,下面将逐一介绍四种单例设计模式: 1.第一种单例设计模式 1.1 代码实现 package com.singleton.one; /**  * 第一种单例设计模式  * @author Administrator  *  */ public class SingleTonOne { // 实例化 private static SingleTonOne instance = new SingleTonOne();

JAVA并发实现四（守护线程和线程阻塞）

守护线程     Java中有两类线程:User Thread(用户线程).Daemon Thread(守护线程) 用户线程即运行在前台的线程,而守护线程是运行在后台的线程. 守护线程作用是为其他前台线程的运行提供便利服务,而且仅在普通.非守护线程仍然运行时才需要,比如垃圾回收线程就是一个守护线程.当VM检测仅剩一个守护线程,而用户线程都已经退出运行时,VM就会退出,因为没有如果没有了被守护者,也就没有继续运行程序的必要了.如果有非守护线程仍然存活,VM就不会退出. 守护线程并非只有虚拟机内部提

Java 中的四种引用及垃圾回收策略

Java 中有四种引用:强引用.软引用.弱引用.虚引用: 其主要区别在于垃圾回收时是否进行回收: 1.强引用 使用最普遍的引用.如果一个对象具有强引用,那就 类似于必不可少的生活用品,垃圾回收器绝不会回收它.当内存空 间不足,Java虚拟机宁愿抛出OutOfMemoryError错误,使程序异常终止,也不会靠随意回收具有强引用的对象来解决内存不足问题. 2.软引用(SoftReference) 如果一个对象只具有软引用,那就类似于可有可物的生活用品.如果内存空间足够,垃圾回收器就不会回收它,如果

总结分析Java常见的四种引用

从JDK1.2版本开始,把对象的引用分为四种级别,从而使程序能更加灵活的控制对象的生命周期.这四种级别由高到低依次为:强引用.软引用.弱引用和虚引用. 1.强引用 本章前文介绍的引用实际上都是强引用,这是使用最普遍的引用.如果一个对象具有强引用,那就 类似于必不可少的生活用品,垃圾回收器绝不会回收它.当内存空 间不足,Java虚拟机宁愿抛出OutOfMemoryError错误,使程序异常终止,也不会靠随意回收具有强引用的对象来解决内存不足问题. 2.软引用(SoftReference) 如果一个

Java并发（四）：volatile的实现原理

synchronized是一个重量级的锁,volatile通常被比喻成轻量级的synchronized volatile是一个变量修饰符,只能用来修饰变量. volatile写:当写一个volatile变量时,JMM会把该线程对应的本地内存中的共享变量刷新到主内存. volatile读:当读一个volatile变量时,JMM会把该线程对应的本地内存置为无效.线程接下来将从主内存中读取共享变量. volatile实现原理 1)JMM把内存屏障指令分为下列四类: StoreLoad Barriers

Java中的四种引用类型比较

1.引用的概念 引用这个概念是与JAVA虚拟机的垃圾回收有关的,不同的引用类型对应不同的垃圾回收策略或时机. 垃圾收集可能是大家感到难于理解的较难的概念之一,因为它并不能总是毫无遗漏地解决Java运行时环境中堆管理的问题. 垃圾回收的大致思路是:当Java虚拟机觉得内存不够用的时候,会触发垃圾回收操作(GC),清除无用的对象,释放内存.可是如何判断一个对象是否是垃圾呢?其中的一个方法是计算指向该对象的引用数量,如果引用数量为0,那么该对象就为垃圾(Thread对象是例外),否则还有用处,不能被回

Java 中的四种引用

1.强引用(Strong Reference)在 Java 中四种引用中是"最强"的,我们平时通过 new 关键字创建的对象都属于强引用,如下面的代码: Person person = new Person();其中的 person 就是一个强引用,只有当它不再被使用后才会被垃圾回收器回收掉.当内存不足时,但是其依然在被使用中,那么垃圾回收器也不会回收其引用的对象:JVM 宁愿报"内存泄漏错误 (OutofMemoryError)",终止程序也不会回收此引用所关联的