CSP 並發編程實戰：JCSP 核心構造與組合模式詳解 (第 2 部分)

前言

承接第 1 部分對 Java 傳統多執行緒痛點的分析，本文正式進入 CSP (Communicating Sequential Processes) 理論與 JCSP (Java CSP) 函式庫的核心實作。重點涵蓋：

1. CSP 基礎概念：Process、Channel、Process Network
2. JCSP 核心 API：CSProcess、ChannelInput/Output、One2OneChannel 等
3. Parallel 組合子構建層級化進程網絡
4. 確定性與非確定性編程：Guard/Alternative 機制
5. 完整範例：IntegrateInt、SquaresInt、ScaleInt/Controller 系統

一、CSP 核心抽象

Process (進程)

• CSP 中「一切皆進程」，封裝資料與演算法，無公開方法 (僅有 run() 入口)。
• 進程擁有獨立控制流 (執行緒)，內部狀態完全私有，消除共享記憶體競爭。
• JCSP 介面：public interface CSProcess { public void run(); } (類似 Runnable 但語義更強)。

Channel (通道)

• 進程間唯一通訊管道，同步、無緩衝 (預設)、點對點。
• 讀寫雙方皆會阻塞，直到對方就緒 (Rendezvous 同步)。
• 四種通道拓撲：
  • One2OneChannel：1 寫 1 讀
  • One2AnyChannel：1 寫多讀 (讀者競爭)
  • Any2OneChannel：多寫 1 讀 (寫者競爭)
  • Any2AnyChannel：多寫多讀
• 支援物件通道 (ChannelInput/Output) 與基本型別通道 (ChannelInputInt/OutputInt 等)，避免裝箱開銷。

Process Network (進程網絡)

• 進程透過通道連接形成網絡，網絡本身可視為一個更大的進程 (組合性)。
• 遞歸組合：Parallel 接受 CSProcess[]，內部進程可再是 Parallel。

二、JCSP 內部實現機制 (關鍵理解)

Channel 零緩衝實現 (ZeroBuffer)

// 寫入端
synchronized void write(Object obj) {
    buffer.put(obj);        // 狀態 EMPTY → FULL
    wait();                 // 釋放鎖，阻塞等待讀取
}

// 讀取端
synchronized Object read() {
    while (buffer.isEmpty()) wait();
    Object obj = buffer.get(); // 狀態 FULL → EMPTY
    notify();                // 喚醒寫入端
    return obj;
}

利用 Java 監視器 (synchronized + wait/notify) 實現 Rendezvous 同步，JCSP 將這些底層細節封裝於 Channel 內部。

Parallel 執行機制

• 建立內部 Barrier (N 方柵欄)。
• 啟動 N-1 個新執行緒跑前 N-1 個 Process，主執行緒跑第 N 個。
• 所有 Process 執行完畢在 Barrier 會合，run() 返回。

三、基礎範例：偶數產生與消費

SendEvenIntsProcess (生產者)

public class SendEvenIntsProcess implements CSProcess {
    private ChannelOutputInt out;
    public SendEvenIntsProcess(ChannelOutputInt out) { this.out = out; }
    public void run() {
        for (int i = 2; i <= 100; i += 2)
            out.write(i);
    }
}

ReadEvenIntsProcess (消費者)

public class ReadEvenIntsProcess implements CSProcess {
    private ChannelInputInt in;
    public ReadEvenIntsProcess(ChannelInputInt in) { this.in = in; }
    public void run() {
        while (true) {
            int val = in.read();
            System.out.println("Read: " + val);
        }
    }
}

DriverProgram (組裝啟動)

public class DriverProgram {
    public static void main(String[] args) {
        One2OneChannelInt chan = new One2OneChannelInt();
        new Parallel(new CSProcess[] {
            new SendEvenIntsProcess(chan.out()),
            new ReadEvenIntsProcess(chan.in())
        }).run();
    }
}

四、進階組合範例：IntegrateInt (累加器網絡)

組件定義

• PlusInt：兩輸入相加 → 輸出
• Delta2Int：單輸入廣播 → 兩輸出
• PrefixInt：預置初始值 (0) → 後續透傳輸入

IntegrateInt 組裝

public class IntegrateInt implements CSProcess {
    private ChannelInputInt in;
    private ChannelOutputInt out;
    public IntegrateInt(ChannelInputInt in, ChannelOutputInt out) {
        this.in = in; this.out = out;
    }
    public void run() {
        One2OneChannelInt a = new One2OneChannelInt();
        One2OneChannelInt b = new One2OneChannelInt();
        One2OneChannelInt c = new One2OneChannelInt();
        new Parallel(new CSProcess[] {
            new PlusInt(in, c, a),      // in + c → a
            new Delta2Int(a, out, b),   // a → out, b
            new PrefixInt(0, b, c)      // 0 → c, then b → c
        }).run();
    }
}

資料流追蹤 (通道 a/b/out)

Channel a: [0, 1, 2, 3, 4, 5, ...]
Channel b: [0, 1, 3, 6, 10, 15, ...]  // 三角數
Channel out: [1, 4, 9, 16, 25, 36, ...] // 平方數！

IntegrateInt 輸出為輸入整數的中間累加和，配合 PrefixInt 初始 0，實現流式累加。

五、網絡中的網絡：SquaresInt (平方數產生器)

public class SquaresInt implements CSProcess {
    private ChannelOutputInt out;
    public SquaresInt(ChannelOutputInt out) { this.out = out; }
    public void run() {
        One2OneChannelInt a = new One2OneChannelInt();
        One2OneChannelInt b = new One2OneChannelInt();
        new Parallel(new CSProcess[] {
            new NumbersInt(a),           // 0,1,2,3... → a
            new IntegrateInt(a, b),      // a 累加 → b (三角數)
            new PairsInt(b, out)         // b 相鄰兩數相加 → out (平方數)
        }).run();
    }
}

數學驗證

第 n 步驟：NumbersInt 輸出 n → IntegrateInt 累加為 n(n-1)/2 → PairsInt 計算 (n-1)(n-2)/2 + n(n-1)/2 = (n-1)²。完美生成平方數序列。

關鍵洞見：SquaresInt 將 IntegrateInt 視為黑盒組件，體現 CSP遞歸組合性。

六、JCSP 確定性特質

• 進程網絡輸出僅取決於輸入，與 JVM 排程策略、多核心執行順序無關。
• 消除了 Heisenbug (難以重現的並發 Bug) 來源。
• 適合安全關鍵、金融交易等需形式化驗證系統。

七、非確定性編程：Guard / Alternative

應用場景

輸出取決於「事件發生順序」而非單純輸入值 (如多路複用、超時控制、優先級搶佔)。

三種選擇策略

三類 Guard

• Channel Guard (AltingChannelInput/Int)：通道有資料可讀時就緒。
• Timer Guard (CSTimer)：達到絕對時間點就緒。
• Skip Guard (Skip)：永遠就緒 (用於非阻塞輪詢)。

八、完整非確定性範例：ScaleInt / Controller 系統

需求

• 放大器每秒接收一整數，乘以當前因子輸出。
• 因子每 5 秒自動翻倍。
• 控制器每 6 秒發送暫停信號，讀取當前因子，計算新因子注入放大器並喚醒。
• 暫停期間放大器直接透傳輸入 (不放大)。

ScaleInt 核心邏輯 (簡化)

public class ScaleInt implements CSProcess {
    private int factor = 2;
    // 正常模式 Guards: suspend(最高), timer, in
    // 暫停模式 Guards: inject(最高), in
    public void run() {
        Alternative normalAlt = new Alternative(new Guard[]{suspend, timer, in});
        Alternative suspendedAlt = new Alternative(new Guard[]{inject, in});
        long timeout = timer.read() + 5000;
        timer.setAlarm(timeout);

        while (true) {
            switch (normalAlt.priSelect()) {
                case 0: // suspend
                    suspend.read(); factor.write(factor);
                    boolean suspended = true;
                    while (suspended) {
                        switch (suspendedAlt.priSelect()) {
                            case 0: factor = inject.read(); suspended = false; timeout = timer.read()+5000; timer.setAlarm(timeout); break;
                            case 1: out.write(in.read()); break; // 透傳
                        }
                    }
                    break;
                case 1: // timer
                    timeout = timer.read()+5000; timer.setAlarm(timeout); factor *= 2; break;
                case 2: // in
                    out.write(factor * in.read()); break;
            }
        }
    }
}

Controller

public class Controller implements CSProcess {
    public void run() {
        CSTimer tim = new CSTimer();
        while (true) {
            tim.after(6000);           // 等待 6 秒
            suspend.write(0);          // 發送暫停
            int curr = factor.read();  // 讀取因子
            curr++;                    // 簡單計算新因子
            inject.write(curr);        // 注入並喚醒
        }
    }
}

DriverProgram 組裝

new Parallel(new CSProcess[] {
    new NumbersInt(temp),                    // 自然數源
    new FixedDelayInt(1000, temp, in),       // 1 秒延遲
    new ScaleInt(2, suspend, in, factor, inject, out),
    new Controller(6000, suspend, inject, factor),
    new PrinterInt(out, "--> ", "\n")
}).run();

九、本部分小結

• 掌握 CSProcess、Channel、Parallel 三大核心構造。
• 理解 Channel 內部 Rendezvous 同步機制 (ZeroBuffer + 監視器)。
• 體驗遞歸組合 : IntegrateInt → SquaresInt → 更大系統。
• 確定性網絡 vs 非確定性選擇 : 兩種並發抽象並存，各適其場景。
• Guard/Alternative 提供聲明式多路複用，優於手寫 select()/wait() 迴圈。

下一部分將探討 CSP 與 AOP 的結構相似性、Barrier/Bucket/CREW 等高級同步構造、ProcessManager 動態管理、JCSP.Net 分布式擴展，以及與 java.util.concurrent 的深度比較。

參考資源

• JCSP 官網 (Kent 大學) / Javadoc / PowerPoint 簡報
• C.A.R. Hoare, Communicating Sequential Processes
• Bill Roscoe, Theory and Practice of Concurrency
• FDR2 模型檢查器 (商業版)
• Peter Welch, Formal Analysis of Concurrent Java Systems

作者簡介

Abhijit Belapurkar，IIT Delhi 計算機科學學士，Infosys 班加羅爾高級技術架構師。