GPars 實戰：在 Groovy 中輕鬆駕馭並發編程

前言：多核時代的並發挑戰

隨著 CPU 核心數從 4、6 核邁向百核、千核，如何榨乾硬體效能成為軟體開發核心課題。傳統 Java 線程模型（synchronized、wait/notify）複雜且易出錯，Groovy、Scala、Clojure 等 JVM 語言推出了更高階的並發抽象。GPars (Groovy Parallel Systems) 將 Actor、Dataflow、Agent、Fork/Join、Parallel Array 等主流並發模型封裝為 Groovy DSL，讓 Java 開發者以近乎原生語法編寫高併發程式。

核心模型速覽

場景一：後台與並行處理——告別串行 I/O 等待

問題：串行呼叫 Twitter API

// langTweets.groovy - 串行版本
@Grab('net.homeip.yusuke:twitter4j:2.0.10')
def recentTweets(api, queryStr) {
    def query = new Query(queryStr)
    query.rpp = 5; query.lang = "en"
    api.search(query).tweets.collect { "[${Thread.currentThread().name}-${queryStr}] @${it.fromUser}: ${it.text}" }
}
def api = new Twitter()
['#erlang','#scala','#clojure'].each { tag ->
    recentTweets(api, tag).each { println it }
}

三個查詢串行執行，總耗時 = Σ(單次查詢耗時)。

解決：GParsExecutorsPool 並行化

// langTweetsParallel.groovy
@Grab('org.codehaus.gpars:gpars:0.10')
import groovyx.gpars.GParsExecutorsPool

GParsExecutorsPool.withPool {
    def retrieveTweets = { query ->
        recentTweets(api, query).each { println it }
    }
    ['#erlang','#scala','#clojure'].each { retrieveTweets.callAsync(it) }
}

關鍵點：

• withPool 建立線程池（預設核心數 + 1）
• callAsync 非阻塞提交任務，返回 Future
• I/O 等待與後續處理並行，總耗時 ≈ max(單次查詢耗時)

線程池客製化

// 指定線程數
GParsExecutorsPool.withPool(8) { ... }

// 自定義 ThreadFactory
def factory = { Runnable r -> 
    def t = new Thread(r); t.name = "worker-${counter++}"; t 
} as ThreadFactory
GParsExecutorsPool.withPool(4, factory) { ... }

場景二：CPU 密集型任務——Fork/Join 與 Parallel Array

分治算法：尋找列表最大值

// computeMax.groovy - 手動 Fork/Join
@Grab('org.codehaus.gpars:gpars:0.10')
import static groovyx.gpars.GParsPool.runForkJoin
import groovyx.gpars.GParsPool

GParsPool.withPool(4) {
    def computedMax = runForkJoin(0, items.size()-1, items) { start, end, list ->
        if (end - start <= 128) return list[start..end].max()
        def mid = (start + end) / 2
        forkOffChild(start, mid, list)
        forkOffChild(mid+1, end, list)
        childrenResults.max()
    }
}

Fork/Join 使用工作竊取雙端佇列，解決傳統 Executor 單一佇列爭用瓶頸：

高階抽象：Parallel Array 一行搞定

// computeMaxPA.groovy
GParsPool.withPool(4) {
    def computedMax = items.parallel.max()  // 內建並行 max()
}

GPars 在 GParsPool.withPool 區塊中為集合注入並行方法：

• collectParallel、findAllParallel、eachParallel、groupByParallel
• 核心函數算子：map、reduce、filter、sum、min、max

實戰：訂單逾期統計

GParsPool.withPool {
    def data = createOrders().parallel
        .filter { it.isLate() }
        .map { it.daysOverdue() }
    println "# overdue = ${data.size()}"
    println "avg overdue = ${data.sum() / data.size()}"
}

聲明式風格，自動並行執行，無需手動分割任務。

場景三：共享狀態管理——三種無鎖範式

Actors：訊息傳遞替代共享記憶體

// rps.groovy - 剪刀石頭布 Actor 版
class Player extends AbstractPooledActor {
    void act() {
        loop { react { reply Move.values()[new Random().nextInt(3)] } }
    }
}
class Coordinator extends AbstractPooledActor {
    void act() {
        loop {
            react { player1.send("play"); player2.send("play") }
            react { msg1 -> react { msg2 -> 
                announce(msg1, msg2)
                if (games-- > 0) send("play") else stop()
            }}
        }
    }
}
[player1, player2, coordinator]*.start()
coordinator << "start"

• react 釋放線程，極輕量（百萬級 Actor 可行）
• 無共享狀態，天然避免競態條件

Agents：異步序列化狀態更新

// zombieApocalypse.groovy
class World { int alive=1000, undead=10; void eatBrains(){...} void shotgun(){...} }
def world = new Agent(new World())

Thread.start { while(!world.val.apocalypse()) { world << { it.eatBrains() }; sleep 200 } }
Thread.start { while(!world.val.apocalypse()) { world << { it.shotgun() }; sleep 200 } }
while(!world.instantVal.apocalypse()) { world.val.report(); sleep 200 }

• 狀態更新透過傳遞函數 { it.eatBrains() } 序列化執行
• 讀取 world.val 獲取不可變快照，無鎖無阻塞

Dataflow：聲明式數據流向

// langTweetsDataflow.groovy
def resultStream = new DataFlowStream()
GParsExecutorsPool.withPool {
    tags.each { tag -> { query -> recentTweets(api, query, resultStream) }.callAsync(tag) }
}
def done = 0
while (done < tags.size()) {
    def msg = resultStream.val  // 阻塞直到有數據
    if (msg == "DONE") done++ else println msg
}

• DataFlowStream 充當單寫單讀通道，「毒丸」訊號標記結束
• Dataflow 變數只能綁定一次，消除競態；死鎖具確定性（測試無死鎖 → 生產無死鎖）

總結與展望

GPars 將 Erlang、Scala、Clojure、Oz 的精華並發模型，以 Groovy DSL 形式統一呈現於 JVM 之上。核心優勢：

• 語法親和力：Java 開發者零門檻上手
• 模型完整：覆蓋任務並行、訊息驅動、狀態管理、數據流
• 效能導向：Fork/Join 針對多核優化，Actor 輕量級支撐大規模併發

未來隨著核心數爆炸增長，聲明式並發（Dataflow、Parallel Array）與無共享架構將成主流。GPars 為 JVM 生態提供了極具實戰價值的探索平台。