java并发的四种风味:thread、executor、forkjoin和actor-亚博电竞手机版
这篇文章讨论了java应用中并行处理的多种方法。从自己管理java线程,到各种更好几的解决方法,executor服务、forkjoin 框架以及计算中的actor模型。
java并发编程的4种风格:threads,executors,forkjoin和actors
我们生活在一个事情并行发生的世界。自然地,我们编写的程序也反映了这个特点,它们可以并发的执行。当然除了python代码(译者注:链接里面讲述了python的全局解释器锁,解释了原因),不过你仍然可以使用jython在jvm上运行你的程序,来利用多处理器电脑的强大能力。
然而,并发程序的复杂程度远远超出了人类大脑的处理能力。相比较而言,我们简直弱爆了:我们生来就不是为了思考多线程程序、评估并发访问有限资源以及预测哪里会发生错误或者瓶颈。
面对这些困难,人类已经总结了不少并发计算的亚博vip888的解决方案和模型。这些模型强调问题的不同部分,当我们实现并行计算时,可以根据问题做出不同的选择。
在这篇文章中,我将会用对同一个问题,用不同的代码来实现并发的亚博vip888的解决方案;然后讨论这些方案有哪些好的地方,有哪些缺陷,可能会有什么样的陷阱在等着你。
我们将介绍下面几种并发处理和异步代码的方式:
• 裸线程
• executors和services
• forkjoin框架和并行流
• actor模型
为了更加有趣一些,我没有仅仅通过一些代码来说明这些方法,而是使用了一个共同的任务,因此每一节中的代码差不多都是等价的。另外,这些代码仅仅是展示用的,初始化的代码并没有写出来,并且它们也不是产品级的软件示例。
对了,最后一件事:在文章最后,有一个小调查,关于你或者你的组织正在使用哪种并发模式。为了你的工程师同胞们,请填一下调查!
任务
任务:实现一个方法,它接收一条消息和一组字符串作为参数,这些字符串与某个搜索引擎的查询页面对应。对每个字符串,这个方法发出一个http请求来查询消息,并返回第一条可用的结果,越快越好。
如果有错误发生,抛出一个异常或者返回空都是可以的。我只是尝试避免为了等待结果而出现无限循环。
简单说明:这次我不会真正深入到多线程如何通讯的细节,或者深入到java内存模型。如果你迫切地想了解这些,你可以看我前面的文章利用jcstress测试并发。
那么,让我们从最直接、最核心的方式来在jvm上实现并发:手动管理裸线程。
方法1:使用“原汁原味”的裸线程
解放你的代码,回归自然,使用裸线程!线程是并发最基本的单元。java线程本质上被映射到操作系统线程,并且每个线程对象对应着一个计算机底层线程。
自然地,jvm管理着线程的生存期,而且只要你不需要线程间通讯,你也不需要关注线程调度。
每个线程有自己的栈空间,它占用了jvm进程空间的指定一部分。
线程的接口相当简明,你只需要提供一个runnable,调用.start()开始计算。没有现成的api来结束线程,你需要自己来实现,通过类似boolean类型的标记来通讯。
在下面的例子中,我们对每个被查询的搜索引擎,创建了一个线程。查询的结果被设置到atomicreference,它不需要锁或者其他机制来保证只出现一次写操作。开始吧!
private static string getfirstresult(string question, listengines) { atomicreference result = new atomicreference<>(); for(string base: engines) { string url = base question; new thread(() -> { result.compareandset(null, ws..get()); }).start(); } while(result.get() == null); // wait for some result to appear return result.get(); }
使用裸线程的主要优点是,你很接近并发计算的操作系统/硬件模型,并且这个模型非常简单。多个线程运行,通过共享内存通讯,就是这样。
自己管理线程的最大劣势是,你很容易过分的关注线程的数量。线程是很昂贵的对象,创建它们需要耗费大量的内存和时间。这是一个矛盾,线程太少,你不能获得良好的并发性;线程太多,将很可能导致内存问题,调度也变得更复杂。
然而,如果你需要一个快速和简单的亚博vip888的解决方案,你绝对可以使用这个方法,不要犹豫。
方法2:认真对待executor和completionservice
另一个选择是使用api来管理一组线程。幸运的是,jvm为我们提供了这样的功能,就是executor接口。executor接口的定义非常简单:
public interface executor { void execute(runnable command); }
它隐藏了如何处理runnable的细节。它仅仅说,“开发者!你只是一袋肉,给我任务,我会处理它!”
更酷的是,executors类提供了一组方法,能够创建拥有完善配置的线程池和executor。我们将使用newfixedthreadpool(),它创建预定义数量的线程,并不允许线程数量超过这个预定义值。这意味着,如果所有的线程都被使用的话,提交的命令将会被放到一个队列中等待;当然这是由executor来管理的。
在它的上层,有executorservice管理executor的生命周期,以及completionservice会抽象掉更多细节,作为已完成任务的队列。得益于此,我们不必担心只会得到第一个结果。
下面service.take()的一次调用将会只返回一个结果。
private static string getfirstresultexecutors(string question, listengines) { executorcompletionservice service = new executorcompletionservice (executors.newfixedthreadpool(4)); for(string base: engines) { string url = base question; service.submit(() -> { return ws..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进程内的所有线程共享。这简化了配置项,因此你不用担心。
private static string getfirstresult(string question, listengines) { // get element as soon as it is available optional result = engines.stream().parallel().map((base) -> { string url = base question; return ws..get(); }).findany(); return result.get(); }
看上面的例子,我们不关心单独的任务在哪里完成,由谁完成。然而,这也意味着,你的应用程序中可能存在一些停滞的任务,而你却无法不知道。在另一篇关于并行流的文章中,我详细地描述了这个问题。并且有一个变通的亚博vip888的解决方案,虽然它并不是世界上最直观的方案。
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模型已经被广泛接受,并被看做是对语言非常有价值的补充。
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..get(); getsender().tell(new result(result), getself()); } else { unhandled(message); } } } static class querier extends untypedactor { private string question; private listengines; private atomicreference result; public querier(string question, list engines, atomicreference 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 engines) { actorsystem system = actorsystem.create("search"); atomicreference 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线程开始,我们逐渐地发现更高级的亚博vip888的解决方案,执行不同的executor服务、forkjoin框架和actor计算模型。
不知道当你面临真实问题时该如何选择?它们都有各自的优缺点,你需要在直观和易用性、配置和增加/减少机器性能等方面做出选择。