Java Socket 編程實戰（三）：TCP 服務端線程池模型與斷點續傳

演進路線

1. 單線程阻塞：一次只能服務一個客戶端
2. 每請求一線程：高併發下線程爆炸、上下文切換開銷大
3. 固定線程池：控制併發數，但無隊列、無擴縮容
4. 成熟線程池：核心/最大線程、工作隊列、拒絕策略、keep-alive、監控

核心協議處理器

public class EchoProtocol implements Runnable {
    private static final int BUFSIZE = 32 * 1024; // 32KB 緩衝
    private final Socket clientSocket;
    private final Logger logger;

    public EchoProtocol(Socket socket, Logger logger) {
        this.clientSocket = socket;
        this.logger = logger;
    }

    public static void handleEchoClient(Socket socket, Logger logger) {
        try (InputStream in = socket.getInputStream();
             OutputStream out = socket.getOutputStream()) {
            byte[] buffer = new byte[BUFSIZE];
            int bytesRead;
            long total = 0;
            while ((bytesRead = in.read(buffer)) != -1) {
                out.write(buffer, 0, bytesRead);
                total += bytesRead;
            }
            logger.info(() -> "Client " + socket.getRemoteSocketAddress() + " echoed " + total + " bytes");
        } catch (IOException e) {
            logger.log(Level.WARNING, "Echo error", e);
        } finally {
            try { socket.close(); } catch (IOException ignored) {}
        }
    }

    @Override public void run() { handleEchoClient(clientSocket, logger); }
}

模型一：每請求一線程 (教學用)

public class TCPEchoServerThread {
    public static void main(String[] args) throws IOException {
        try (ServerSocket server = new ServerSocket(5500)) {
            Logger logger = Logger.getLogger("practical");
            while (true) {
                Socket client = server.accept();
                new Thread(new EchoProtocol(client, logger)).start();
                logger.info("Started thread for " + client.getRemoteSocketAddress());
            }
        }
    }
}

模型二：固定線程池 (手寫簡易版)

public class TCPEchoServerPool {
    public static void main(String[] args) throws IOException {
        int poolSize = 3;
        try (ServerSocket server = new ServerSocket(5500)) {
            Logger logger = Logger.getLogger("practical");
            for (int i = 0; i < poolSize; i++) {
                new Thread(() -> {
                    while (true) {
                        try {
                            Socket client = server.accept(); // 多線程競爭 accept
                            EchoProtocol.handleEchoClient(client, logger);
                        } catch (IOException e) {
                            logger.log(Level.WARNING, "Accept failed", e);
                        }
                    }
                }, "Worker-" + i).start();
            }
        }
    }
}

缺點：無隊列緩衝、無動態調整、異常處理簡陋。

模型三：生產級線程池 (基於 java.util.concurrent)

封裝 TaskExecutor

public class ThreadPoolTaskExecutor {
    private static final ExecutorService executor = 
        new ThreadPoolExecutor(
            4,                      // corePoolSize
            16,                     // maximumPoolSize
            60L, TimeUnit.SECONDS,  // keepAliveTime
            new LinkedBlockingQueue<>(1024), // workQueue
            new ThreadFactoryBuilder().setNameFormat("tcp-worker-%d").build(),
            new ThreadPoolExecutor.AbortPolicy() // 拒絕策略
        );

    // 同步提交 + 超時控制
    public static <T> T submit(Callable<T> task, long timeout, TimeUnit unit) 
            throws TimeoutException, ExecutionException, InterruptedException {
        Future<T> future = executor.submit(task);
        return future.get(timeout, unit);
    }

    public static void submit(Runnable task) { executor.submit(task); }

    public static void shutdown() { executor.shutdown(); }
}

服務端主循環

public class TCPEchoServerExecutor {
    public static void main(String[] args) throws IOException {
        try (ServerSocket server = new ServerSocket(5500)) {
            Logger logger = Logger.getLogger("practical");
            while (true) {
                Socket client = server.accept();
                // 異步提交，主線程立即返回 accept 下一個
                ThreadPoolTaskExecutor.submit(new EchoProtocol(client, logger));
            }
        } finally {
            ThreadPoolTaskExecutor.shutdown();
        }
    }
}

斷點續傳 / 大文件傳輸擴展

協議設計

// 客戶端發送：文件元數據 (JSON) + 文件內容
// 元數據：{fileName, fileSize, fileMD5, chunkIndex, totalChunks, offset}
// 服務端：RandomAccessFile.seek(offset) 寫入對應分片
// 完成後校驗 MD5，回覆成功/失敗

關鍵點

• RandomAccessFile 支持任意位置讀寫，實現分片追加
• 分片索引 chunkIndex 確保亂序到達也能正確組裝
• 臨時文件命名：filename.part，完成後重命名
• 進度持久化：Redis/DB 記錄已收分片，支持服務重啟恢復
• 超時清理：定時掃描超過 24h 未完成的臨時文件並刪除

Netty 升級建議

手寫 NIO/線程池易出 Bug，生產環境強烈建議直接使用 Netty：

• EventLoopGroup 完美解決線程模型
• ChannelPipeline 責任鏈處理編解碼、心跳、SSL、流控
• ByteBuf 零拷貝、引用計數、內存池
• 內建斷點續傳示例 FileServer / FileClient

總結

TCP 服務端線程模型演進：單線程 → 每請求一線程 → 固定池 → 成熟線程池 → Netty。核心竅門：

1. 用 ThreadPoolExecutor 替代手寫池，獲得隊列、擴縮容、監控、拒絕策略
2. 協議層面設計分片元數據，配合 RandomAccessFile 實現斷點續傳
3. 超過 1 萬併發、需 SSL、心跳、流控時，無腦上 Netty