深入了解Java语言中的并发性选项有何不同

前言

Java™ 工程师在努力让并发性容易为开发人员所用。尽管做了不少的改进,但并发性仍然是 Java 平台的一个复杂、容易出错的部分。一部分复杂之处在于理解语言本身中的并发性的低级抽象,这些抽象在您的代码中填满了同步的代码块。另一个复杂之处来自一些新库,比如 fork/join,这些库在某些场景中非常有用,但在其他场景中收效甚微。了解容易混乱的大量低级选项需要专业经验和时间。

脱离 Java 语言的优势之一是,能够改善和简化并发性等区域。每种 Java 下一代语言都为此问题提供了独特的答案,利用了该语言的默认编程风格。在本期文章中,我首先将会介绍函数式编程风格的优势:轻松并行化。我会深入分析 Scala 和 Groovy 的细节(下一期文章将全面介绍 Clojure)。然后介绍 Scala actor。

完美数

数学家尼科马库斯(诞生于公元前 6 世纪)将自然数分为惟一的完美数(perfect number)、过剩数(abundant number) 或亏数(deficient number)。一个完美数等于它的正因数(不包括它本身)之和。例如,6 是一个完美数,因为它的因数是 1、2、3 和 6,28 也是完美数 (28 = 1 + 2 + 4 + 7 + 14)。过剩数的因素之和大于该数,亏数的因数之和小于该数。

这里使用完美数分类法是为了方便介绍。除非要处理大量数字,是否查找因素对于从并行化中获益而言是一个微不足道的问题。使用更多线程可带来一些益处,但线程之间的切换开销对细粒度的作业而言代价很高。

让现有代码并行化

在 “函数式编码风格” 那一期的文章中,我们鼓励您使用更高级的抽象,比如化简、映射和过滤器,而不是迭代。此方法的优势之一是容易并行化。

我的 函数式思维 系列的读者熟悉包含完美数 的数字分类模式(参见 完美数 边栏)。我在该系列中展示的任何解决方案都没有利用并发性。但是因为这些解决方案使用了转换函数,比如 map,所以我可以在每种 Java.net 语言中做极少的工作来创建并行化的版本。

清单 1 是完美数分类器的一个 Scala 示例。

清单 1. Scala 中的并行完美数分类器

object NumberClassifier {
def isFactor(factor: Int, number: Int) =
number % factor == 0
def factors(number: Int) = {
val factorsBelowSqrt = (1 to Math.sqrt(number).toInt).par.filter (isFactor(_, number))
val factorsAboveSqrt = factorsBelowSqrt.par.map(number / _)
(factorsBelowSqrt ++ factorsAboveSqrt).toList.distinct
}
def sum(factors: Seq[Int]) =
factors.par.foldLeft(0)(_ + _)
def isPerfect(number: Int) =
sum(factors(number)) - number == number
}

清单 1 中的 factors() 方法返回一个数的因数列表,使用 isFactor() 方法过滤所有可能的值。factors() 方法使用了我在 “函数式思维:转换和优化” 中更详细地介绍的一种优化。简单来讲,过滤每个数来查找因素的效率很低,因为根据定义,一个因数是其乘积等于目标数的两个数之一。

相反,我仅过滤不超过目标数的平方根的数,然后通过将目标数除以每个小于平方根的因数来生成对称因数列表。在 清单 1 中,factorsBelowSqrt 变量包含过滤操作的结果。factorsAboveSqrt 的值是现有列表的映射,用于生成这些对称值。最后,factors() 的返回值是一个串联的列表,它从一个并行的List 转换为常规的 List。

请注意,清单 1 中添加了 par 修饰符。该修饰符会导致 filter、map 和 foldLeft 并行运行,从而能够使用多个线程来处理请求。par 方法(在整个 Scala 集合库中都是一致的)将该序列转换为并行序列。因为两种类型的序列反映了它们的签名,所以 par 函数变成了并行化某个操作的临时方式。

在 Scala 中并行化常见问题的简单性,在语言设计和函数模式上都经过证实。函数式编程鼓励使用通用的函数,比如 map、filter 和 reduce,运行时以不可见的方式可以进一步优化它们。Scala 语言设计人员考虑到了这些优化,最终产生了集合 API 的设计。

边缘情况

在 清单 1 的 factors() 方法实现中,整数的平方根(例如,16 的平方根:4)显示在两个列表中。因此,factors() 方法返回的最后一行是对 distinct 函数的调用,它从列表中删除了重复值。您也可以在每一处都使用 Set,而不是只在列表中使用它,但 List 常常拥有 Set 中所没有的有用函数。

Groovy 也允许轻松地修改现有的函数代码,通过 GPars 库让它并行化,该库捆绑在各个 Groovy 发行版中。GPars 框架在内置的 Java 并行性原语之上创建有用的抽象,常常将它们包装在语法糖中。GPars 提供了令人眼花缭乱的并行机制,其中一种机制可用于分配线程池,然后将操作分布到这些池中。清单 2 中给出了一个使用 Groovy 编写的,使用 GPars 线程池的完美数分类器。

清单 2. Groovy 中的并行完美数分类器

class NumberClassifierPar {
static def factors(number) {
GParsPool.withPool {
def factors = (1..round(sqrt(number) + 1)).findAllParallel { number % it == 0 }
(factors + factors.collectParallel { number / it }).unique()
}
}
static def sumFactors(number) {
factors(number).inject(0, { i, j -> i + j })
}
static def isPerfect(number) {
sumFactors(number) - number == number
}
}

清单 2 中的 factors() 方法使用了与 清单 1 相同的算法:它生成不超过目标数的平方根的所有因数,然后生成剩余的因数并返回串联的集合。与 清单 1 中一样,我使用 unique() 方法来确保整数的平方根不会生成重复值。

无需像 Scala 中一样放大集合来创建对称并行版本,Groovy 的设计人员创建了该语言的转换方法的 xxxParallel() 版本(例如 findAllParallel() 和 collectParallel())。但除非这些方法包装在 GPars 线程池代码块中,否则它们不会起作用。

在 清单 2 中,我创建了一个线程池,调用 GParsPool.withPool 创建一个代码块,支持在该代码块中使用 xxxParallel() 方法。withPool 方法存在其他变体。例如,您可指定池中的线程数量。

Clojure 通过 化简器 库提供了一种类似的临时并行化机制。使用转换函数的化简器版本来实现自动并行化,例如,
使用 r/map 代替 map。(r/ 是化简器命名空间。)化简器的实现是 Clojure 的语法灵活性中的一个引人注目的案例分析,它通过极小的更改实现了强大的添加功能。

Scala 中的 actor

Scala 包含众多并发性和并行性机制。一种较流行的机制是 actor 模型,它提供了将工作分布到线程上的优势,而没有同步的复杂性。在概念上,actor 有能力完成工作,然后将一个非阻塞的结果发送给协调器。要创建一个 actor,需要创建 Actor 类的子类并实现 act() 方法。通过使用 Scala 的语法糖,可绕过许多定义仪式,在代码块内定义 actor。

我没有为 清单 1 中的数字分类器执行的一种优化是,使用线程对作业的因数查找部分进行分区。如果我的计算机上有 4 个处理器,我可为每个处理器创建一个线程并拆分工作。例如,如果我尝试找到数字 16 的因数之和,那么我可以安排处理器 1 来查找从 1 到 4 的因数(并求和),安排处理器 2 来处理 5 到 8,依此类推。使用 actor 是一种自然的选择:我为每个范围创建了一个 actor,独立地执行每个 actor(通过语法糖隐式执行或通过调用它的 act() 方法来显式执行),然后收集结果,如清单 3 所示。

清单 3. 使用 Scala 中的 actor 识别完美数

object NumberClassifier extends App {
def isPerfect(candidate: Int) = {
val RANGE = 10000
val numberOFPartitions = (candidate.toDouble / RANGE).ceil.toInt
val coordinator = self
for (i <- 0 until numberOFPartitions) {
val lower = i * RANGE + 1
val upper = candidate.min((i + 1) * RANGE)
actor {
var partialSum = 0
for (j <- lower to upper)
if (candidate % j == 0) partialSum += j
coordinator ! partialSum
}
}
var responsesExpected = numberOFPartitions
var sum = 0
while (responsesExpected > 0) {
receive {
case partialSum : Int =>
responsesExpected -= 1
sum += partialSum
}
}
sum == 2 * candidate
}
}

为了保持此示例的简单性,我将 isPerfect() 编写为单个完整的函数。我首先基于常量 RANGE 创建了一些分区。其次,我需要一种方式来收集 actor 所生成的消息。在 coordinator 变量中,我有一个引用可供 actor 向其发送消息,其中 self 是 Actor 的一个成员,表示 Scala 中获取线程标识符的可靠方式。

我然后为分区编号创建一个循环,使用 RANGE 偏移来生成范围的下限和上限。接下来,为该范围创建一个 actor,使用 Scala 的语法糖来避免正式的类定义。在 actor 内,我为 partialSum 创建了一个临时保存器,然后分析该范围,将找到的因数收集到 partialSum 中。收集部分和(此范围内的所有因数的和)后, (coordinator ! partialSum) 向协调器发回一条消息,使用感叹号运算符。(这种消息传递语法的灵感来源于 Erlang 语言,用作一种对另一个线程执行非阻塞调用的途径。)

接下来,我启动了一个循环,等待所有 actor 完成处理。在等待过程中,我进入了一个 receive 代码块。在该代码块内,我想要一条 Int 消息,我在本地将它分配给 partialSum,然后递减想要的响应数量,将该部分添加到总和中。所有 actor 完成且报告结果后,该方法的最后一行将该和与候选数的 2 倍相比较。如果比较结果为 true,那么我的候选数就是一个完美数,该函数的返回值为 true。

actor 的一个不错的优势是所有权分区。每个 actor 都有一个 partialSum 局部变量,但它们从不彼此联系。通过协调器收到消息时,底层执行机制是不可见的:您创建了一个 receive 块,其他实现细节是不可见的。

Scala 中的 actor 机制是 Java 下一代语言封装 JVM 的现有工具并使用一致的抽象来扩展它们的优秀示例。用 Java 语言编写类似的代码,并使用低级并发性原语,这些操作都需要非常复杂地协调多个线程。Scala 中的 actor 隐藏了所有复杂性,留下的是容易理解的抽象。

结束语

Java 下一代语言都为 Java 语言中的并发性难题提供了答案,而且每种语言以不同方式解决了这些问题。在本期文章中,我演示了所有三种 Java 下一代语言如何实现临时并行化。我还演示了 Scala 中的 actor 模型,构建了一个数字分类器来并行计算因数之和。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

时间: 2019-06-12

浅谈Java并发中的内存模型

什么是JavaMemoryModel(JMM)? JMM通过构建一个统一的内存模型来屏蔽掉不同硬件平台和不同操作系统之间的差异,让Java开发者无需关注不同平台之间的差异,达到一次编译,随处运行的目的,这也正是Java的设计目的之一. CPU和内存 在讲JMM之前,我想先和大家聊聊硬件层面的东西.大家应该都知道执行运算操作的CPU本身是不具备存储能力的,它只负责根据指令对传递进来的数据做相应的运算,而数据存储这一任务则交给内存去完成.虽然内存的运行速度虽然比起硬盘快非常多,但是和3GHZ,4GH

通俗易懂学习java并发工具类-Semaphore,Exchanger

1. 控制资源并发访问--Semaphore Semaphore可以理解为信号量,用于控制资源能够被并发访问的线程数量,以保证多个线程能够合理的使用特定资源. Semaphore就相当于一个许可证,线程需要先通过acquire方法获取该许可证,该线程才能继续往下执行,否则只能在该方法出阻塞等待.当执行完业务功能后,需要通过release()方法将许可证归还,以便其他线程能够获得许可证继续执行. Semaphore可以用于做流量控制,特别是公共资源有限的应用场景,比如数据库连接.假如有多个线程读取

java并发之AtomicInteger源码分析

问题 (1)什么是原子操作? (2)原子操作和数据库的ACID有啥关系? (3)AtomicInteger是怎么实现原子操作的? (4)AtomicInteger是有什么缺点? 简介 AtomicInteger是java并发包下面提供的原子类,主要操作的是int类型的整型,通过调用底层Unsafe的CAS等方法实现原子操作. 还记得Unsafe吗?点击链接直达[java Unsafe详细解析] 原子操作 原子操作是指不会被线程调度机制打断的操作,这种操作一旦开始,就一直运行到结束,中间不会有任何

浅谈java并发之计数器CountDownLatch

CountDownLatch简介 CountDownLatch顾名思义,count + down + latch = 计数 + 减 + 门闩(这么拆分也是便于记忆=_=) 可以理解这个东西就是个计数器,只能减不能加,同时它还有个门闩的作用,当计数器不为0时,门闩是锁着的:当计数器减到0时,门闩就打开了. 如果你感到懵比的话,可以类比考生考试交卷,考生交一份试卷,计数器就减一.直到考生都交了试卷(计数器为0),监考老师(一个或多个)才能离开考场.至于考生是否做完试卷,监考老师并不关注.只要都交了试

java并发编程实例分析

java并发编程是java程序设计语言的一块重点,在大部分的业务场景中都需要并发编程. 比如:并发的去处理http请求,这样就可以使得一台机器同时处理多个请求,大大提高业务的响应效率,从而使用用户体验更加流畅. java如何并发编程,要注意以下几个方面: 1.java语言中的多线程操作:创建和启动线程的几种方式. 2.共享变量的同步问题,要保证线程安全,辨别哪些变量是线程安全的.那些变量是线程不安全的,对于不安全的变量我们要想办法让其同步,一般也就是加锁. 3.线程锁:包括方法锁和synchro

浅谈Java并发 J.U.C之AQS:CLH同步队列

CLH同步队列是一个FIFO双向队列,AQS依赖它来完成同步状态的管理,当前线程如果获取同步状态失败时,AQS则会将当前线程已经等待状态等信息构造成一个节点(Node)并将其加入到CLH同步队列,同时会阻塞当前线程,当同步状态释放时,会把首节点唤醒(公平锁),使其再次尝试获取同步状态. 在CLH同步队列中,一个节点表示一个线程,它保存着线程的引用(thread).状态(waitStatus).前驱节点(prev).后继节点(next),其定义如下: static final class Node

Java并发计数器的深入理解

前言 一提到线程安全的并发计数器,AtomicLong 必然是第一个被联想到的工具.Atomic* 一系列的原子类以及它们背后的 CAS 无锁算法,常常是高性能,高并发的代名词.本文将会阐释,在并发场景下,使用 AtomicLong 来充当并发计数器将会是一个糟糕的设计,实际上存在不少 AtomicLong 之外的计数器方案.近期我研究了一些 Jdk1.8 以及 JCTools 的优化方案,并将它们的对比与实现细节整理于此. 阅读本文前 本文相关的基准测试代码均可在博主的 github 中找到,

Java并发编程之Semaphore(信号量)详解及实例

Java并发编程之Semaphore(信号量)详解及实例 概述 通常情况下,可能有多个线程同时访问数目很少的资源,如客户端建立了若干个线程同时访问同一数据库,这势必会造成服务端资源被耗尽的地步,那么怎样能够有效的来控制不可预知的接入量呢?及在同一时刻只能获得指定数目的数据库连接,在JDK1.5 java.util.concurrent 包中引入了Semaphore(信号量),信号量是在简单上锁的基础上实现的,相当于能令线程安全执行,并初始化为可用资源个数的计数器,通常用于限制可以访问某些资源(物

Java并发编程Semaphore计数信号量详解

Semaphore 是一个计数信号量,它的本质是一个共享锁.信号量维护了一个信号量许可集.线程可以通过调用acquire()来获取信号量的许可:当信号量中有可用的许可时,线程能获取该许可:否则线程必须等待,直到有可用的许可为止. 线程可以通过release()来释放它所持有的信号量许可(用完信号量之后必须释放,不然其他线程可能会无法获取信号量). 简单示例: package me.socketthread; import java.util.concurrent.ExecutorService;

Java 并发编程:volatile的使用及其原理解析

Java并发编程系列[未完]: •Java 并发编程:核心理论 •Java并发编程:Synchronized及其实现原理 •Java并发编程:Synchronized底层优化(轻量级锁.偏向锁) •Java 并发编程:线程间的协作(wait/notify/sleep/yield/join) •Java 并发编程:volatile的使用及其原理 一.volatile的作用 在<Java并发编程:核心理论>一文中,我们已经提到过可见性.有序性及原子性问题,通常情况下我们可以通过Synchroniz

Java并发编程:CountDownLatch与CyclicBarrier和Semaphore的实例详解

Java并发编程:CountDownLatch与CyclicBarrier和Semaphore的实例详解 在java 1.5中,提供了一些非常有用的辅助类来帮助我们进行并发编程,比如CountDownLatch,CyclicBarrier和Semaphore,今天我们就来学习一下这三个辅助类的用法. 以下是本文目录大纲: 一.CountDownLatch用法 二.CyclicBarrier用法 三.Semaphore用法 若有不正之处请多多谅解,并欢迎批评指正. 一.CountDownLatch

Java并发编程(CyclicBarrier)实例详解

Java并发编程(CyclicBarrier)实例详解 前言: 使用JAVA编写并发程序的时候,我们需要仔细去思考一下并发流程的控制,如何让各个线程之间协作完成某项工作.有时候,我们启动N个线程去做一件事情,只有当这N个线程都达到某一个临界点的时候,我们才能继续下面的工作,就是说如果这N个线程中的某一个线程先到达预先定义好的临界点,它必须等待其他N-1线程也到达这个临界点,接下来的工作才能继续,只要这N个线程中有1个线程没有到达所谓的临界点,其他线程就算抢先到达了临界点,也只能等待,只有所有这N

Java 并发编程学习笔记之Synchronized简介

一.Synchronized的基本使用 Synchronized是Java中解决并发问题的一种最常用的方法,也是最简单的一种方法.Synchronized的作用主要有三个:(1)确保线程互斥的访问同步代码(2)保证共享变量的修改能够及时可见(3)有效解决重排序问题.从语法上讲,Synchronized总共有三种用法: (1)修饰普通方法 (2)修饰静态方法 (3)修饰代码块 接下来我就通过几个例子程序来说明一下这三种使用方式(为了便于比较,三段代码除了Synchronized的使用方式不同以外,

Java 并发编程学习笔记之核心理论基础

并发编程是Java程序员最重要的技能之一,也是最难掌握的一种技能.它要求编程者对计算机最底层的运作原理有深刻的理解,同时要求编程者逻辑清晰.思维缜密,这样才能写出高效.安全.可靠的多线程并发程序.本系列会从线程间协调的方式(wait.notify.notifyAll).Synchronized及Volatile的本质入手,详细解释JDK为我们提供的每种并发工具和底层实现机制.在此基础上,我们会进一步分析java.util.concurrent包的工具类,包括其使用方式.实现源码及其背后的原理.本

java并发编程之cas详解

CAS(Compare and swap)比较和替换是设计并发算法时用到的一种技术.简单来说,比较和替换是使用一个期望值和一个变量的当前值进行比较,如果当前变量的值与我们期望的值相等,就使用一个新值替换当前变量的值.这听起来可能有一点复杂但是实际上你理解之后发现很简单,接下来,让我们跟深入的了解一下这项技术. CAS的使用场景 在程序和算法中一个经常出现的模式就是"check and act"模式.先检查后操作模式发生在代码中首先检查一个变量的值,然后再基于这个值做一些操作.下面是一个

JAVA并发编程有界缓存的实现详解

JAVA并发编程有界缓存的实现 1.有界缓存的基类 package cn.xf.cp.ch14; /** * *功能:有界缓存实现基类 *时间:下午2:20:00 *文件:BaseBoundedBuffer.java *@author Administrator * * @param <V> */ public class BaseBoundedBuffer<V> { private final V[] buf; private int tail; private int head

Java并发编程示例(十):线程组

对线程分组是Java并发API提供的一个有趣功能.我们可以将一组线程看成一个独立单元,并且可以随意操纵线程组中的线程对象.比如,可以控制一组线程来运行同样的任务,无需关心有多少线程还在运行,还可以使用一次中断调用中断所有线程的执行. Java提供了ThreadGroup类来控制一个线程组.一个线程组可以通过线程对象来创建,也可以由其他线程组来创建,生成一个树形结构的线程. 根据<Effective Java>的说明,不再建议使用ThreadGroup.建议使用Executor. --D瓜哥特此