JDK 1.0/1.1 僅提供 wait()/notify() 等基礎監視器方法，缺乏高層同步構件。開發者需自行構建信號量、管道等核心並發原語。雖然 Java 5+ 引入 java.util.concurrent (Semaphore, PipedInputStream/OutputStream 等)，但理解其底層實現原理、掌握自定義同步工具的編寫技巧，對於深度掌握並發編程、解決特殊場景問題、面試核心考點至關重要。本文深度解析基於 wait/notify 的計數信號量實現、JDK 管道流缺陷分析與自製管道流改進。
public class MySemaphore {
private int permits; // 許可證計數
public MySemaphore(int initialPermits) {
this.permits = initialPermits;
}
// P 操作：獲取許可
public synchronized void acquire() throws InterruptedException {
permits--;
if (permits < 0) {
wait(); // 釋放監視器鎖，進入等待集合
}
}
// V 操作：釋放許可
public synchronized void release() {
permits++;
if (permits <= 0) {
notify(); // 喚醒一個等待執行緒 (隨機/實現相關)
}
}
// 非阻塞嘗試獲取
public synchronized boolean tryAcquire() {
if (permits > 0) {
permits--;
return true;
}
return false;
}
// 可用許可數 (近似值，並發下瞬間快照)
public synchronized int availablePermits() {
return Math.max(permits, 0);
}
}
acquire/release 整體原子執行，避免檢查-修改-阻塞的競態條件。release()。notify() 足夠；多許可或公平性要求時建議 notifyAll() 避免信號丟失 (極罕見但理論可能)。wait() 拋出 InterruptedException，調用者需處理 (重新中斷/拋出/轉換)。
MySemaphore pool = new MySemaphore(10); // 10 個資料庫連線
// 獲取連線
pool.acquire();
Connection conn = createConnection();
try { /* 使用連線 */ } finally {
closeConnection(conn);
pool.release(); // 歸還許可
}
PipedInputStream / PipedOutputStreamconnect() 後不可斷開重連。close() 會永久標記 closed=true、清空緩衝區、斷開連接引用，**無法重置狀態重新連接**。PIPE_SIZE = 1024 為 private static final，不可配置。大數據傳輸易阻塞寫入端。wait/notifyAll 在共享緩衝區協調，無獨立鎖，擴展性差。
public class MyPipedInputStream extends InputStream {
protected byte[] buffer;
protected int in = -1; // 寫入指針
protected int out = 0; // 讀取指針
protected boolean closedByReader = false;
protected boolean closed = false;
protected boolean connected = false;
protected MyPipedOutputStream sink; // 連接的輸出流
public static final int DEFAULT_PIPE_SIZE = 1024;
private final int pipeSize;
public MyPipedInputStream(int pipeSize) {
this.pipeSize = pipeSize > 0 ? pipeSize : DEFAULT_PIPE_SIZE;
this.buffer = new byte[this.pipeSize];
}
public MyPipedInputStream() { this(DEFAULT_PIPE_SIZE); }
// 連接輸出流
public synchronized void connect(MyPipedOutputStream src) throws IOException {
if (connected || src.connected) throw new IOException("Already connected");
this.sink = src;
src.sink = this;
connected = true;
src.connected = true;
}
@Override
public synchronized int read() throws IOException {
if (!connected) throw new IOException("Pipe not connected");
while (in == out && !closedByReader && !closed) {
try { wait(); } catch (InterruptedException e) { throw new IOException("Interrupted"); }
}
if (closedByReader || closed) return -1;
int b = buffer[out] & 0xFF;
out = (out + 1) % pipeSize;
notifyAll(); // 喚醒寫入端
return b;
}
@Override
public synchronized int read(byte[] b, int off, int len) throws IOException {
// 批量讀取實現... (類似單字節邏輯，循環填充)
}
// 關鍵改進：close 重置狀態，支援重新連接
@Override
public synchronized void close() throws IOException {
in = -1; out = 0;
closedByReader = true;
connected = false;
closed = true;
buffer = new byte[pipeSize]; // 重新分配緩衝區
if (sink != null) {
sink.sink = null;
sink.connected = false;
sink = null;
}
notifyAll();
}
// 供輸出流調用：寫入資料
synchronized void receive(int b) throws IOException {
while (in == ((out - 1 + pipeSize) % pipeSize) && !closedByReader && !closed) {
try { wait(); } catch (InterruptedException e) { throw new IOException("Interrupted"); }
}
if (closedByReader || closed) throw new IOException("Pipe closed");
buffer[in] = (byte) b;
in = (in + 1) % pipeSize;
notifyAll();
}
synchronized void receive(byte[] b, int off, int len) throws IOException {
for (int i = 0; i < len; i++) receive(b[off + i]);
}
synchronized void receivedLast() {
closed = true;
notifyAll();
}
}
public class MyPipedOutputStream extends OutputStream {
private MyPipedInputStream sink;
private boolean closed = false;
private boolean connected = false;
public void connect(MyPipedInputStream snk) throws IOException {
snk.connect(this); // 委託輸入流完成雙向綁定
}
@Override
public void write(int b) throws IOException {
if (!connected || sink == null) throw new IOException("Pipe not connected");
sink.receive(b);
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
if (!connected || sink == null) throw new IOException("Pipe not connected");
sink.receive(b, off, len);
}
@Override
public void close() throws IOException {
if (sink != null) {
sink.receivedLast(); // 通知讀取端資料結束
sink.closed = true;
}
sink = null;
connected = false;
closed = true;
}
// 重置狀態支援重連 (需配合新的 InputStream)
public void reset() {
sink = null;
connected = false;
closed = false;
}
}
MyPipedInputStream in = new MyPipedInputStream(64 * 1024); // 64KB 緩衝區
MyPipedOutputStream out = new MyPipedOutputStream();
out.connect(in);
// 生產者執行緒
new Thread(() -> {
try {
for (int i = 0; i < 1000; i++) out.write(("Data " + i + "\n").getBytes());
out.close();
} catch (IOException e) { e.printStackTrace(); }
}).start();
// 消費者執行緒 (主執行緒)
try (BufferedReader br = new BufferedReader(new InputStreamReader(in))) {
String line;
while ((line = br.readLine()) != null) System.out.println("Received: " + line);
} catch (IOException e) { e.printStackTrace(); }
// 完成後可重置並重新連接新流 (舊流已 GC)
out.reset();
in = new MyPipedInputStream(64 * 1024);
out.connect(in);
// ... 第二輪傳輸 ...
| 需求 | JDK 5+ 標準方案 | 優勢 |
|---|---|---|
| 信號量 | java.util.concurrent.Semaphore | 公平/非公平模式、可中斷、超時獲取、許可數動態調整 |
| 管道流 | java.io.PipedInputStream/OutputStream (改進版) | JDK 9+ 修復了重連問題，緩衝區仍固定 |
| 高效執行緒間傳輸 | LinkedBlockingQueue, ArrayBlockingQueue | 無序列化開銷、支援背壓、豐富 API (poll, offer, take, drainTo) |
| 資料流處理 | Reactive Streams (Flow API, Reactor, RxJava) | 背壓、操作符豐富、函數式組合 |
