您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 程式設計入門

Java 5 並發編程實戰：線程池、阻塞隊列、ReentrantLock 與 Callable/Future 打造高效網路伺服器

編輯：Java 程式設計入門

Java 5 並發包 (JSR-166) 核心組件

Executor 框架：ExecutorService、ThreadPoolExecutor、Executors 工廠
同步器：ReentrantLock、Condition、Semaphore、CountDownLatch、CyclicBarrier
併發集合：ConcurrentHashMap、CopyOnWriteArrayList、BlockingQueue 實作 (ArrayBlockingQueue/LinkedBlockingQueue)
任務抽象：Callable<V>、Future<V>、FutureTask
原子變數：AtomicInteger、AtomicReference、LongAdder (JDK 8)

傳統模型 vs 線程池模型

維度	每請求一線程	線程池模型
建立/銷毀開銷	高 (頻繁 GC、OS 切換)	低 (線程復用)
資源可控性	差 (可能耗盡記憶體/檔案描述符)	強 (核心/最大線程數、隊列長度可配置)
回應延遲	含線程建立時間	僅含隊列等待 + 執行時間
適用場景	長連接、低併發	短請求、高併發、突發流量

實戰：基於線程池的 Echo Server

package com.example.concurrent;

import java.io.*;
import java.net.*;
import java.util.concurrent.*;

public class ThreadPoolEchoServer {
    private static final int PORT = 19527;
    private static final int POOL_SIZE = 10;

    private final ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
    private final ServerSocket serverSocket;

    public ThreadPoolEchoServer() throws IOException {
        serverSocket = new ServerSocket(PORT);
        serverSocket.setReuseAddress(true);
    }

    public void start() {
        System.out.println("Server listening on " + PORT);
        try {
            while (!serverSocket.isClosed()) {
                Socket client = serverSocket.accept();
                pool.execute(new ClientHandler(client));
            }
        } catch (IOException e) {
            if (!serverSocket.isClosed()) e.printStackTrace();
        } finally {
            shutdown();
        }
    }

    public void shutdown() {
        try { serverSocket.close(); } catch (IOException ignore) {}
        pool.shutdown();
        try { pool.awaitTermination(60, TimeUnit.SECONDS); } 
        catch (InterruptedException e) { pool.shutdownNow(); }
    }

    private static class ClientHandler implements Runnable {
        private final Socket socket;
        private static final AtomicInteger counter = new AtomicInteger();
        private static final ReentrantLock lock = new ReentrantLock();

        ClientHandler(Socket socket) { this.socket = socket; }

        @Override
        public void run() {
            int id = incrementAndGet();
            try (DataOutputStream out = new DataOutputStream(socket.getOutputStream());
                 BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()))) {
                
                String welcome = "Hello, client #" + id + "\n";
                out.write(welcome.getBytes());
                out.flush();

                String line;
                while ((line = in.readLine()) != null) {
                    if ("quit".equalsIgnoreCase(line)) break;
                    out.write(("Echo: " + line + "\n").getBytes());
                    out.flush();
                }
            } catch (IOException e) {
                System.err.println("Client #" + id + " error: " + e.getMessage());
            } finally {
                try { socket.close(); } catch (IOException ignore) {}
            }
        }

        private int incrementAndGet() {
            lock.lock();
            try { return counter.incrementAndGet(); } finally { lock.unlock(); }
        }
    }

    public static void main(String[] args) throws IOException {
        new ThreadPoolEchoServer().start();
    }
}

關鍵技術點解析

1. 固定大小線程池

ExecutorService pool = Executors.newFixedThreadPool(10);
// 等價於
new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS,
                       new LinkedBlockingQueue<Runnable>());

核心=最大=10，無閒置超時，無界隊列。生產環境建議自行構建 ThreadPoolExecutor 設定有界隊列、拒絕策略、線程工廠命名。

2. ReentrantLock 替代 synchronized

可中斷鎖獲取：lock.lockInterruptibly()
嘗試非阻塞獲取：lock.tryLock()、tryLock(timeout, unit)
公平鎖選項：new ReentrantLock(true)
多條件變數：lock.newCondition() 實現多等待隊列
效能：高爭用下優於 synchronized (JDK 6+ biased locking 縮小差距)

3. Callable + Future 非同步有回傳任務

ExecutorService es = Executors.newCachedThreadPool();
Future<String> future = es.submit(() -> {
    // 耗時運算/遠端呼叫
    return "result";
});
// 其他工作...
String result = future.get(); // 阻塞等待結果

配合 CompletionService 批次提交、按完成順序取結果。

4. 泛型確保型別安全

ExecutorService、BlockingQueue<Task>、Future<Result> 全鏈路泛型，編譯期捕獲型別錯誤。

生產級增強清單

優雅關閉：shutdown() + awaitTermination() + shutdownNow() 兜底
監控指標：pool.getActiveCount()、getQueue().size()、getCompletedTaskCount() 暴露給 Prometheus
線程命名：自訂 ThreadFactory 加入業務前綴 (如 echo-server-worker-1)，便於 jstack 排查
拒絕策略：CallerRunsPolicy (回調呼叫者執行)、DiscardOldestPolicy、自訂記錄告警
異常處理：Thread.setDefaultUncaughtExceptionHandler 或 Future.get() 捕獲 ExecutionException