GPars是Groovy（和Java）中用于高級(jí)并發(fā)的開(kāi)源庫(kù)。如果您聽(tīng)過(guò)actors、dataflow或parallel collections之類(lèi)的術(shù)語(yǔ)，并且希望用Java友好的語(yǔ)言嘗試這些概念，那么現(xiàn)在您有機(jī)會(huì)了。在本文中，我計(jì)劃對(duì)gpar中可用的抽象進(jìn)行快速概述。然后我們將更詳細(xì)地研究基于流的并發(fā)數(shù)據(jù)處理、并行化收集操作以及異步運(yùn)行組合函數(shù)。

隨著多核芯片逐漸成為主流電腦、平板電腦和手機(jī)的標(biāo)準(zhǔn)，并發(fā)編程越來(lái)越重要。不幸的是，我們從Java中了解到的廣泛使用的基于線程的并發(fā)模型與人腦的工作方式并不匹配。線程和鎖在代碼中引入了太多的不確定性，這常常導(dǎo)致難以追蹤和修復(fù)的細(xì)微錯(cuò)誤。這樣的代碼不能被可靠地測(cè)試或分析。不可避免地，為了使并發(fā)編程有效，我們需要使用更自然的心智模型。

并發(fā)可以是Groovy

將直觀的并發(fā)模型引入主流是gpar雄心勃勃的挑戰(zhàn)。我們采用了眾所周知的并發(fā)概念，如actors、CSP、dataflow等，并用Java實(shí)現(xiàn)了它們，使用了一種美味的Groovy DSL，使庫(kù)具有流暢和易于使用的風(fēng)格。盡管主要針對(duì)Groovy，但一些gpar抽象可以直接從Java使用。由于Groovy社區(qū)對(duì)并發(fā)的興趣以及他們對(duì)項(xiàng)目的支持，gpar目前是Groovy發(fā)行版的標(biāo)準(zhǔn)部分。在Groovy中啟動(dòng)并運(yùn)行并發(fā)不需要額外的設(shè)置。

被認(rèn)為是致命的循環(huán)

我想讓你停下來(lái)想一想計(jì)算機(jī)科學(xué)專(zhuān)業(yè)的學(xué)生在學(xué)習(xí)編程時(shí)通常被分配的一些瑣碎的練習(xí)。例如，這樣的任務(wù)之一是查找集合的最大值?？赡軙?huì)應(yīng)用一些復(fù)雜的度量來(lái)增加解決方案的計(jì)算量。你首先想到的是什么算法？

你很有可能會(huì)提出一個(gè)迭代來(lái)遍歷這個(gè)集合并記住迄今為止發(fā)現(xiàn)的最大元素。一旦我們到達(dá)集合的末尾，我們記住的迄今為止最大的元素必須是整個(gè)集合的全局最大值。清楚，簡(jiǎn)單-好吧，錯(cuò)了！如果你想知道原因，就繼續(xù)讀下去。

做你的選擇

在并發(fā)空間中似乎沒(méi)有一個(gè)一刀切的解決方案。多種范式已經(jīng)逐漸出現(xiàn)，盡管有一些重疊，但它們都適用于不同類(lèi)型的問(wèn)題。GPars本身在2008年作為一個(gè)用于并行收集處理的小型Groovy庫(kù)起步。不久之后，它增加了對(duì)actor模型的支持。隨著時(shí)間的推移，其他并發(fā)抽象已經(jīng)被集成。以下是GPars版本0.12中當(dāng)前可用內(nèi)容的快速列表：

• 并行集合提供了直觀的方式來(lái)并行處理Java / Groovy集合，地圖和一般所有幾何可分解問(wèn)題的處理

• 異步功能使Groovy閉包能夠異步運(yùn)行，同時(shí)又可以毫不費(fèi)力地協(xié)調(diào)它們之間的相互通信。

• Fork / Join使您能夠同時(shí)處理遞歸的分治式算法

• 數(shù)據(jù)流變量（又名Promises）為線程間通信提供了一種輕量級(jí)的機(jī)制。

• 數(shù)據(jù)流通道和運(yùn)算符使您可以將活動(dòng)的數(shù)據(jù)轉(zhuǎn)換元素組織到高度并發(fā)的數(shù)據(jù)處理管道和網(wǎng)絡(luò)中

• CSP是基于理論數(shù)學(xué)的著名并發(fā)模型，它使用通過(guò)同步通道進(jìn)行通信的獨(dú)立并發(fā)運(yùn)行流程的抽象

• Actor / Active對(duì)象為您提供了低禮儀事件驅(qū)動(dòng)的主動(dòng)組件的抽象，這些組件異步交換數(shù)據(jù)

• 顧名思義，代理可以保護(hù)多個(gè)線程需要同時(shí)訪問(wèn)的數(shù)據(jù)

您可以在GPars用戶指南中查看每個(gè)模型的詳細(xì)信息，并將它們與它們的典型適用范圍進(jìn)行并排比較。另外，Dierk?nig即將出版的第二版書(shū)“Groovy in Action”對(duì)gpar進(jìn)行了詳細(xì)的介紹。

在本文中，我選擇了三種最有可能向您展示直觀并發(fā)好處的抽象：并行集合、異步函數(shù)和數(shù)據(jù)流操作符。我們就潛進(jìn)去吧！

幾何分解

現(xiàn)在，這是一個(gè)很好的地方來(lái)解釋為什么我們前面描述的求最大值的順序算法是一個(gè)錯(cuò)誤的選擇。并不是說(shuō)這個(gè)解決方案是錯(cuò)誤的。很明顯，它給了你正確的答案，不是嗎？它失敗的地方在于它的有效性。它禁止隨著工人人數(shù)的增加而擴(kuò)大規(guī)模。它完全忽略了這樣一種可能性，即系統(tǒng)可能會(huì)在這個(gè)問(wèn)題上放置多個(gè)處理器。

超市幾十年前就解決了這個(gè)難題。當(dāng)收銀臺(tái)的隊(duì)伍排得太長(zhǎng)時(shí)，他們會(huì)另外叫一個(gè)出納來(lái)為顧客服務(wù)，這樣工作量就會(huì)分散，吞吐量也會(huì)增加。

回到尋找最大值的問(wèn)題：利用Groovy函數(shù)集合API，GPars添加了每個(gè)流行迭代方法的并行版本，如eachParallel（）、collectpallel（）、findAllParallel（）、maxpallel（）和其他方法。這些方法對(duì)用戶隱藏了實(shí)際的實(shí)現(xiàn)。在幕后，集合被劃分成更小的塊，可能是按層次組織的，每個(gè)塊都將由不同的線程處理（圖1）。

實(shí)際工作由用戶必須提供的線程池中的線程執(zhí)行。GPAR提供了兩種類(lèi)型的線程池：

• GParsExecutorsPool 使用直接的Java 5 Executor

• GParsPool 使用Fork / Join線程池

Parallel collections in use:GParsPool.withPool 16, {    def myFavorite = programmingLanguages.collectParallel {it.grabSpec()}                        .maxParallel {it.spec.count(/parallel/)}}

在withPool代碼塊中，并行收集方法自動(dòng)在周?chē)€程池的線程之間分配工作。向池中添加的線程越多，獲得的并行度就越高。如果沒(méi)有明確的池大小要求，池將為運(yùn)行時(shí)檢測(cè)到的每個(gè)可用處理器創(chuàng)建一個(gè)線程，從而提供最大的計(jì)算能力。這樣就不會(huì)有人為的上限限制算法內(nèi)部的并行性。無(wú)論是在舊的單處理器機(jī)器上還是在未來(lái)的百核芯片上，代碼都將全速運(yùn)行。

GPars并行集合API提供了經(jīng)常被稱(chēng)為循環(huán)并行問(wèn)題或更一般的幾何分解問(wèn)題的解決方案。還有其他類(lèi)型的挑戰(zhàn)需要更具創(chuàng)造性的并發(fā)方法。我們將在文章的下一部分討論其中的兩個(gè)。

異步函數(shù)

在看到集合被并發(fā)處理之后，我們現(xiàn)在將重點(diǎn)討論函數(shù)。Groovy對(duì)函數(shù)編程有很好的支持；畢竟，能夠并行化方法和函數(shù)調(diào)用肯定會(huì)派上用場(chǎng)。為了接近Groovy所在的領(lǐng)域，我選擇了軟件項(xiàng)目構(gòu)建編排問(wèn)題作為下一步的主題。

注意：當(dāng)并行化構(gòu)建進(jìn)程時(shí)，通常I/O限制比CPU限制更多，我們明顯地增加了磁盤(pán)和網(wǎng)絡(luò)帶寬的利用率，而不是處理器的利用率。它不僅是CPU，而且是其他資源，通過(guò)并發(fā)代碼可以提高它們的利用率。顯然，所演示的原理可以完全以同樣的方式應(yīng)用于CPU受限的問(wèn)題

讓我們假設(shè)我們有一組函數(shù)，可能實(shí)現(xiàn)為shell腳本、gradle任務(wù)或GAnt方法，它們可以執(zhí)行構(gòu)建的不同部分。傳統(tǒng)的構(gòu)建腳本可以如下所示：

清單1：構(gòu)建腳本的順序版本

println "Starting the build process."def projectRoot = checkout('git@github.com:vaclav/GPars.git')def classes = compileSources(projectRoot)def api = generateAPIDoc(projectRoot)def guide = generateUserDocumentation(projectRoot)def result = deploy(packageProject(classes, api, guide))

我們現(xiàn)在的任務(wù)是盡可能地安全地并行化構(gòu)建，而不需要過(guò)度的努力。您可能會(huì)看到compileSources（）、generateAPIDoc（）和generateUserGuide（）可以安全地并行運(yùn)行，因?yàn)樗鼈儧](méi)有相互依賴關(guān)系。他們只需要等待checkout（）完成就可以開(kāi)始工作。但是，腳本會(huì)連續(xù)運(yùn)行它們。

我相信你能想象出比這個(gè)更復(fù)雜的構(gòu)建場(chǎng)景。然而，如果沒(méi)有一個(gè)好的抽象，如果我們的任務(wù)是并發(fā)地運(yùn)行構(gòu)建任務(wù)，即使使用這樣一個(gè)人工簡(jiǎn)化的構(gòu)建腳本，我們也要做相當(dāng)多的工作（清單2）。你覺(jué)得這個(gè)怎么樣？

清單2：使用異步函數(shù)的構(gòu)建腳本的并發(fā)版本

withPool {    /* We need asynchronous variants of all the individual build steps */    def aCheckout = checkout.asyncFun()    def aCompileSources = compileSources.asyncFun()    def aGenerateAPIDoc = generateAPIDoc.asyncFun()    def aGenerateUserDocumentation = generateUserDocumentation.asyncFun()    def aPackageProject = packageProject.asyncFun()    def aDeploy = deploy.asyncFun()    /* Here's the composition of asynchronous build steps to form a process */    Promise projectRoot = aCheckout('git@github.com:vaclav/GPars.git')    Promise classes = aCompileSources(projectRoot)    Promise api = aGenerateAPIDoc(projectRoot)    Promise guide = aGenerateUserDocumentation(projectRoot)    Promise result = aDeploy(aPackageProject(classes, api, guide))    /* Now we're setup and can wait for the build to finish */    println "Starting the build process. This line is quite likely to be printed first ..."    println result.get()}

線路保持完全相同。我們只通過(guò)asyncFun（）方法將原始函數(shù)轉(zhuǎn)換為異步函數(shù)。另外，整個(gè)代碼塊現(xiàn)在被包裝在GParsPool.with池（）塊，以便函數(shù)有一些線程用于它們的艱苦工作。

顯然，在asyncFun（）方法中發(fā)生了一些神奇的事情，允許函數(shù)異步運(yùn)行，并且在它們需要彼此的數(shù)據(jù)時(shí)進(jìn)行協(xié)作。

細(xì)節(jié)之美

本質(zhì)上，asyncFun（）將原始函數(shù)包裝到新函數(shù)中。新函數(shù)的簽名與原始函數(shù)的簽名略有不同。例如，原始的compileSources（）函數(shù)將字符串作為參數(shù)并返回字符串作為結(jié)果： 

String compileSources = {String projectRoot -> ...}

新構(gòu)造的aCompileSources（）函數(shù)返回字符串的承諾，而不是字符串本身。此外，String和Promise

PromiseaCompileSources = {String | PromiseprojectRoot -> ...}

信守諾言

Promise接口是幾個(gè)GPars api的基礎(chǔ)。這有點(diǎn)類(lèi)似于java.util.concurrent文件.Future，因?yàn)樗硎疽粋€(gè)正在進(jìn)行的異步活動(dòng)，可以通過(guò)其blocking get（）方法等待和獲取結(jié)果。兩者之間最重要的區(qū)別是，與Java的未來(lái)不同，承諾允許非阻塞讀取。

promise.then {println "Now we have a result: $it"}

這允許我們的異步函數(shù)僅在計(jì)算所需的所有值都可用時(shí)使用系統(tǒng)線程。因此，例如，只有在類(lèi)、api和guide局部變量都綁定到結(jié)果值之后，打包才會(huì)開(kāi)始。在此之前，aPackage（）函數(shù)將在后臺(tái)靜默地等待，沒(méi)有計(jì)劃也沒(méi)有活動(dòng)（圖2）。

現(xiàn)在，您應(yīng)該能夠看到構(gòu)建塊是多么完美地結(jié)合在一起。由于異步函數(shù)返回并接受承諾，因此它們可以按照與同步原始函數(shù)相同的方式進(jìn)行組合。函數(shù)組合的第二個(gè)也是可能更突出的好處是，我們不必明確指定哪些任務(wù)可以并行運(yùn)行。我敢肯定，如果我們繼續(xù)將任務(wù)添加到構(gòu)建過(guò)程中，我們很快就會(huì)失去關(guān)于哪些活動(dòng)可以安全地與哪些其他活動(dòng)并行運(yùn)行的全局視圖。幸運(yùn)的是，我們的異步函數(shù)將在運(yùn)行時(shí)自行發(fā)現(xiàn)并行性。在任何時(shí)候，所有準(zhǔn)備好參數(shù)的任務(wù)都將從分配的線程池中獲取一個(gè)線程并開(kāi)始運(yùn)行。通過(guò)限制池中的線程數(shù)，可以設(shè)置并行運(yùn)行的任務(wù)數(shù)的上限。

數(shù)據(jù)流在哪里

現(xiàn)在，我們已經(jīng)準(zhǔn)備好進(jìn)行今天的集合中最有趣的抽象—數(shù)據(jù)流并發(fā)。在上一個(gè)示例的基礎(chǔ)上，我們現(xiàn)在繼續(xù)。為了構(gòu)建多個(gè)項(xiàng)目，我們將反復(fù)運(yùn)行構(gòu)建腳本。如果您愿意，可以將其視為未來(lái)構(gòu)建服務(wù)器的初始階段。各種項(xiàng)目的構(gòu)建請(qǐng)求將通過(guò)管道傳入，我們的構(gòu)建服務(wù)器將在系統(tǒng)資源允許的情況下依次處理它們。

您可能會(huì)嘗試最簡(jiǎn)單的解決方案—為每個(gè)傳入請(qǐng)求運(yùn)行上一個(gè)練習(xí)中基于異步函數(shù)的代碼。然而，這在很大程度上是次優(yōu)的可能性很高。由于請(qǐng)求隊(duì)列中堆積了多個(gè)請(qǐng)求，我們有機(jī)會(huì)獲得更大的并行性。不僅同一項(xiàng)目的獨(dú)立部分可以并行構(gòu)建，而且不同項(xiàng)目的不同部分也可以同時(shí)處理。簡(jiǎn)單地說(shuō)，不同項(xiàng)目的處理可以在時(shí)間上重疊（圖3）。

順其自然

這種模型自然適合于此類(lèi)問(wèn)題，稱(chēng)為數(shù)據(jù)流網(wǎng)絡(luò)。它通常用于并行數(shù)據(jù)處理，如加密和壓縮、數(shù)據(jù)挖掘、圖像處理等?；旧?，數(shù)據(jù)流網(wǎng)絡(luò)由活動(dòng)的數(shù)據(jù)轉(zhuǎn)換元素（稱(chēng)為操作符）組成，通過(guò)異步通道連接。每個(gè)操作符使用來(lái)自其輸入通道的數(shù)據(jù)，并通過(guò)多個(gè)輸出通道發(fā)布結(jié)果。它還有一個(gè)相關(guān)的轉(zhuǎn)換功能，它將通過(guò)輸入通道接收的數(shù)據(jù)轉(zhuǎn)換為向下發(fā)送到輸出通道的數(shù)據(jù)。在這種情況下，操作符共享一個(gè)線程池，因此沒(méi)有數(shù)據(jù)可處理的非活動(dòng)操作符不會(huì)占用系統(tǒng)線程。對(duì)于我們的簡(jiǎn)單構(gòu)建服務(wù)器，網(wǎng)絡(luò)可以如圖4所示。

我們每一步都有一個(gè)操作員。這些通道表示構(gòu)建任務(wù)之間的依賴關(guān)系，每個(gè)操作符只需要一個(gè)系統(tǒng)線程，并在其所有輸入通道都有值可讀取之后啟動(dòng)計(jì)算。

清單3：使用數(shù)據(jù)流操作符的并發(fā)構(gòu)建服務(wù)器

/* We need channels to wire active elements together */def urls = new DataflowQueue()def checkedOutProjects = new DataflowBroadcast()def compiledProjects = new DataflowQueue()def apiDocs = new DataflowQueue()def userDocs = new DataflowQueue()def packages = new DataflowQueue()def done = new DataflowQueue()/* Here's the composition of individual build steps into a process */operator(inputs: [urls], outputs: [checkedOutProjects], maxForks: 3) {url ->bindAllOutputs checkout(url)}operator([checkedOutProjects.createReadChannel()],         [compiledProjects]) {projectRoot ->bindOutput compileSources(projectRoot)}operator(checkedOutProjects.createReadChannel(), apiDocs) {projectRoot ->bindOutput generateAPIDoc(projectRoot)}operator(checkedOutProjects.createReadChannel(), userDocs) {projectRoot ->bindOutput generateUserDocumentation(projectRoot)}operator([compiledProjects, apiDocs, userDocs],         [packages]) {classes, api, guide ->bindOutput packageProject(classes, api, guide)}def deployer = operator(packages, done) {packagedProject ->if (deploy(packagedProject) == 'success') bindOutput true    else bindOutput false}/* Now we're setup and can wait for the build to finish */println "Starting the build process. This line is quite likely to be printed first ..."deployer.join()  //Wait for the last operator in the network to finish

這個(gè)模型應(yīng)用于我們的問(wèn)題的好處是，例如，當(dāng)?shù)谝粋€(gè)操作符執(zhí)行項(xiàng)目簽出并獲取項(xiàng)目的源代碼時(shí)，它可以立即獲取隊(duì)列中的下一個(gè)請(qǐng)求，并在編譯、打包和部署前很久就開(kāi)始獲取其源代碼。

通過(guò)更改分配給網(wǎng)絡(luò)中特定任務(wù)的操作員數(shù)量，可以調(diào)整系統(tǒng)。例如，如果我們意識(shí)到獲取源是一個(gè)瓶頸，并且假設(shè)硬件（網(wǎng)絡(luò)帶寬）仍然沒(méi)有得到充分利用，那么我們可以通過(guò)增加源獲取操作符的數(shù)量來(lái)提高服務(wù)器的吞吐量（圖5）。

顯然，您的調(diào)優(yōu)選項(xiàng)比分叉大量使用的操作符要遠(yuǎn)得多。為了簡(jiǎn)單地說(shuō)出一些其他的可能性，您可以考慮在網(wǎng)絡(luò)的重復(fù)部分之間采用負(fù)載平衡方案，實(shí)施生產(chǎn)節(jié)流——可以通過(guò)同步通信通道，也可以使用類(lèi)似看板的工作進(jìn)度節(jié)流方案。對(duì)于某些問(wèn)題，可以考慮數(shù)據(jù)緩存和推測(cè)性計(jì)算。

摘要

在深入了解Groovy并發(fā)之后，您可能處于一個(gè)很好的位置來(lái)進(jìn)行更深入的研究。為了快速入門(mén)，請(qǐng)考慮遵循gpar的一個(gè)快速方法，我絕對(duì)建議您查看用戶指南以了解更多詳細(xì)信息。

如果你帶著GPAR去兜風(fēng)我會(huì)很高興的。去享受并發(fā)性吧，因?yàn)椴l(fā)性太棒了！