您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 併發編程技巧：信號量實現與管道流增強版實戰

編輯：Java 相關知識

Java 基礎類庫未直接提供信號量類（JDK 1.5+ 引入 java.util.concurrent.Semaphore），早期開發者需自行基于 wait/notify 實現。同時，JDK 自帶 PipedInputStream/OutputStream 存在關閉後無法重連、緩衝區固定 1KB 等缺陷。本文展示基于內置鎖的計數信號量實現、自定義可重連/可配置緩衝區管道流實現，並分析其原理與適用場景。

一、計數信號量實現

信號量維護一組許可證，acquire() 獲取許可（無則阻塞），release() 歸還許可（喚醒等待者）。經典應用：資源池控制（數據庫連接、線程池）、生產者-消費者協作。

基于 synchronized + wait/notify 的實現


public class SimpleSemaphore {
    private int permits; // 可用許可數
    
    public SimpleSemaphore(int initialPermits) {
        if (initialPermits < 0) throw new IllegalArgumentException();
        this.permits = initialPermits;
    }
    
    // 獲取一個許可，阻塞直到可用
    public synchronized void acquire() throws InterruptedException {
        while (permits == 0) {
            wait(); // 釋放鎖並等待
        }
        permits--;
    }
    
    // 歸還一個許可，喚醒一個等待者
    public synchronized void release() {
        permits++;
        notify(); // 喚醒一個等待線程（notifyAll 更安全）
    }
    
    // 可選：嘗試獲取，帶超時
    public synchronized boolean tryAcquire(long timeoutMs) throws InterruptedException {
        long deadline = System.currentTimeMillis() + timeoutMs;
        while (permits == 0) {
            long remaining = deadline - System.currentTimeMillis();
            if (remaining <= 0) return false;
            wait(remaining);
        }
        permits--;
        return true;
    }
}

關鍵點解析

synchronized 保證 permits 讀寫、wait/notify 原子性。
while(permits == 0) wait() 防禦虛假喚醒。
notify() 單喚醒足夠（每次釋放一個許可），但 notifyAll() 更健壯防信號丟失。
生產者-消費者模式：信號量初始化為 0，生產者 release()，消費者 acquire()，實現高效協作（避免輪詢）。

現代替代：JDK 1.5+ `java.util.concurrent.Semaphore`

支持公平/非公平模式、批量獲取/釋放、可中斷獲取、超時獲取、許可數查詢等豐富功能。
底層基于 AQS (AbstractQueuedSynchronizer) 實現，性能優於內置鎖版本。
生產環境強烈建議直接使用 JUC Semaphore。

二、增強版管道流：可重連、可配置緩衝區

JDK PipedInputStream/OutputStream 缺陷：

close() 後內部狀態不可重置，無法再次 connect()。
緩衝區固定 1024 字節，不可配置，高吞吐易阻塞。

自定義實現核心思路

繼承 InputStream/OutputStream，內部維護字節數組緩衝區、讀寫指標、狀態標誌。
connect() 建立雙向引用；close() 重置所有狀態（緩衝區、指標、連接標誌、關閉標誌），支持重複連接。
緩衝區大小通過構造器參數配置（如 64KB）。
讀寫操作使用 synchronized(this) 同步，配合 wait/notifyAll 實現阻塞協作。

關鍵 close() 實現（核心修復點）


// MyPipedInputStream
@Override public void close() throws IOException {
    synchronized (this) {
        in = -1; out = 0;           // 重置讀寫指標
        closedByReader = true;
        connected = false;          // 關鍵：標記為未連接，允許再次 connect()
        closed = true;
        buffer = new byte[PIPE_SIZE]; // 重新分配緩衝區（可選，或保留複用）
        notifyAll();                // 喚醒可能阻塞在寫端的線程
    }
}

// MyPipedOutputStream
@Override public void close() throws IOException {
    synchronized (this) {
        if (sink != null) {
            sink.receivedLast();    // 通知讀端流結束
            sink.closed = true;
        }
        sink = null;                // 斷開引用
        connected = false;          // 關鍵：允許重連
        notifyAll();
    }
}

讀寫協作邏輯（簡化）


// 寫入
public void write(byte[] b, int off, int len) throws IOException {
    synchronized (this) {
        while (len > 0) {
            while (availableSpace() == 0) { // 緩衝區滿
                if (sink.closedByReader) throw new IOException("Pipe closed");
                wait(); // 等待讀取騰出空間
            }
            int writeLen = Math.min(len, availableSpace());
            System.arraycopy(b, off, buffer, in, writeLen);
            in = (in + writeLen) % buffer.length;
            len -= writeLen; off += writeLen;
            notifyAll(); // 喚醒讀端
        }
    }
}

// 讀取
public int read(byte[] b, int off, int len) throws IOException {
    synchronized (this) {
        while (availableData() == 0) {
            if (closedByWriter && in == out) return -1; // EOF
            wait(); // 等待數據寫入
        }
        int readLen = Math.min(len, availableData());
        // ... 從 buffer 複製到 b ...
        out = (out + readLen) % buffer.length;
        notifyAll(); // 喚醒寫端
        return readLen;
    }
}

三、對比總結與選型建議

組件	手寫簡易版	JUC 官方版 / 成熟庫
信號量	教學理解原理、極簡依賴環境	生產首選：`java.util.concurrent.Semaphore`、`LongAdder` 計數器
管道流	特殊需求：可重連、大緩衝區、無依賴	生產首選：`BlockingQueue` (內存)、`Disruptor` (高性能)、Netty/NIO Channel (網絡)