Java Socket 网络编程与线程池优化方案

2.2 TCP 三次握手与四次挥手

在建立 TCP 连接时，需要经过三次握手：

1. 客户端发送 SYN 包到服务器
2. 服务器响应 SYN-ACK 包
3. 客户端发送 ACK 包确认

断开连接时的四次挥手：

1. 主动方发送 FIN 包
2. 被动方响应 ACK 包
3. 被动方发送 FIN 包
4. 主动方响应 ACK 包

三、多线程处理 Socket 连接

3.1 为什么需要多线程？

单线程服务器一次只能处理一个客户端连接，这在实际应用中是不可接受的。为了支持并发连接，我们需要引入多线程机制。

3.2 子线程处理 Socket 连接

public class MultiThreadServer {
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(8888);
        System.out.println("多线程服务器启动...");
        while (true) {
            // 接受客户端连接
            Socket clientSocket = serverSocket.accept();
            System.out.println("新客户端连接：" + clientSocket.getInetAddress());
            // 为每个客户端创建一个新线程
            ClientHandler handler = new ClientHandler(clientSocket);
            Thread thread = new Thread(handler);
            thread.start();
        }
    }
}

class ClientHandler implements Runnable {
    private Socket socket;

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

    @Override
    public void run() {
        try {
            // 处理客户端请求
            InputStream is = socket.getInputStream();
            OutputStream os = socket.getOutputStream();
            // 模拟业务处理
            Thread.sleep(1000);
            // 读取请求
            byte[] buffer = new byte[1024];
            int len = is.read(buffer);
            String request = new String(buffer, 0, len);
            // 处理请求并响应
            String response = "处理结果：" + request;
            os.write(response.getBytes());
            os.flush();
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

3.3 使用 Callable 和 Future 获取处理结果

public class CallableServer {
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(8888);
        ExecutorService executor = Executors.newCachedThreadPool();
        while (true) {
            Socket clientSocket = serverSocket.accept();
            Callable<String> task = new CallableClientHandler(clientSocket);
            Future<String> future = executor.submit(task);
            // 可以异步获取处理结果
            executor.execute(() -> {
                try {
                    String result = future.get();
                    System.out.println("任务处理结果：" + result);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

class CallableClientHandler implements Callable<String> {
    private Socket socket;

    public CallableClientHandler(Socket socket) {
        this.socket = socket;
    }

    @Override
    public String call() throws Exception {
        // 处理客户端请求
        InputStream is = socket.getInputStream();
        byte[] buffer = new byte[1024];
        int len = is.read(buffer);
        String request = new String(buffer, 0, len);
        // 模拟复杂处理
        Thread.sleep(2000);
        socket.close();
        return "处理完成：" + request;
    }
}

四、传统多线程模型的缺点

4.1 资源消耗问题

public class ResourceTest {
    public static void main(String[] args) {
        // 测试线程创建的资源消耗
        long startTime = System.currentTimeMillis();
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            Thread thread = new Thread(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            threads.add(thread);
            thread.start();
        }
        // 等待所有线程完成
        threads.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        long endTime = System.currentTimeMillis();
        System.out.println("创建 1000 个线程耗时：" + (endTime - startTime) + "ms");
        // 查看内存使用情况
        Runtime runtime = Runtime.getRuntime();
        System.out.println("最大内存：" + runtime.maxMemory() / 1024 / 1024 + "MB");
        System.out.println("已分配内存：" + runtime.totalMemory() / 1024 / 1024 + "MB");
        System.out.println("剩余内存：" + runtime.freeMemory() / 1024 / 1024 + "MB");
    }
}

4.2 传统多线程的主要缺点

1. 线程创建和销毁开销大
   • 每个线程需要约 1MB 的栈内存
   • 创建线程消耗 CPU 时间（约 0.5-1ms）
   • 销毁线程同样需要系统资源
2. 线程管理复杂
   • 难以控制并发线程数量
   • 线程间通信复杂
   • 异常处理困难
3. 稳定性问题
   • 线程过多导致内存溢出
   • 上下文切换频繁，CPU 利用率下降
   • 系统资源耗尽风险
4. 可扩展性差
   • 硬编码线程数量
   • 难以根据系统负载动态调整

五、线程池优化方案

5.1 线程池的核心优势

public class ThreadPoolAdvantages {
    public static void main(String[] args) {
        // 1. 降低资源消耗：复用已创建的线程
        // 2. 提高响应速度：任务到达时，线程已存在
        // 3. 提高线程的可管理性：统一分配、调优和监控
        // 4. 提供更多功能：定时执行、定期执行等
        ExecutorService executor = Executors.newFixedThreadPool(10);
        // 监控线程池状态
        monitorThreadPool((ThreadPoolExecutor) executor);
    }

    private static void monitorThreadPool(ThreadPoolExecutor executor) {
        new Thread(() -> {
            while (true) {
                System.out.println("=== 线程池状态监控 ===");
                System.out.println("核心线程数：" + executor.getCorePoolSize());
                System.out.println("活动线程数：" + executor.getActiveCount());
                System.out.println("最大线程数：" + executor.getMaximumPoolSize());
                System.out.println("队列大小：" + executor.getQueue().size());
                System.out.println("已完成任务数：" + executor.getCompletedTaskCount());
                System.out.println("========================");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    break;
                }
            }
        }).start();
    }
}

5.2 线程池优化 Socket 服务器

public class OptimizedSocketServer {
    // 自定义线程工厂，便于识别线程来源
    private static class ServerThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        ServerThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            namePrefix = "socket-pool-" + poolNumber.getAndIncrement() + "-thread-";
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            // 设置为非守护线程
            t.setDaemon(false);
            // 设置优先级
            t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

    // 自定义拒绝策略
    private static class ServerRejectedExecutionHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            // 记录日志
            System.err.println("任务被拒绝：" + r.toString());
            // 尝试重新放入队列
            if (!executor.isShutdown()) {
                try {
                    executor.getQueue().put(r);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    public static void main(String[] args) throws IOException {
        // 创建自定义线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            5, // 核心线程数
            20, // 最大线程数
            60L, // 空闲线程存活时间
            TimeUnit.SECONDS, // 时间单位
            new LinkedBlockingQueue<>(100), // 任务队列
            new ServerThreadFactory(), // 线程工厂
            new ServerRejectedExecutionHandler() // 拒绝策略
        );
        // 允许核心线程超时退出
        executor.allowCoreThreadTimeOut(true);
        ServerSocket serverSocket = new ServerSocket(8888);
        System.out.println("优化后的 Socket 服务器启动...");
        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("服务器关闭中...");
            try {
                executor.shutdown();
                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                    executor.shutdownNow();
                }
                serverSocket.close();
                System.out.println("服务器已关闭");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }));
        // 监控线程池
        startThreadPoolMonitor(executor);
        // 接收客户端连接
        while (!executor.isShutdown()) {
            try {
                Socket clientSocket = serverSocket.accept();
                executor.execute(new OptimizedClientHandler(clientSocket, executor));
            } catch (IOException e) {
                if (!executor.isShutdown()) {
                    System.err.println("接收连接错误：" + e.getMessage());
                }
            }
        }
    }

    private static void startThreadPoolMonitor(ThreadPoolExecutor executor) {
        ScheduledExecutorService monitor = Executors.newSingleThreadScheduledExecutor();
        monitor.scheduleAtFixedRate(() -> {
            System.out.println("\n=== 线程池监控报告 ===");
            System.out.println(String.format("活动线程数：%d/%d", executor.getActiveCount(), executor.getMaximumPoolSize()));
            System.out.println("队列大小：" + executor.getQueue().size());
            System.out.println("已完成任务：" + executor.getCompletedTaskCount());
            System.out.println("========================");
        }, 0, 5, TimeUnit.SECONDS);
    }
}

class OptimizedClientHandler implements Runnable {
    private Socket socket;
    private ThreadPoolExecutor executor;

    public OptimizedClientHandler(Socket socket, ThreadPoolExecutor executor) {
        this.socket = socket;
        this.executor = executor;
    }

    @Override
    public void run() {
        String clientInfo = socket.getInetAddress() + ":" + socket.getPort();
        System.out.println("开始处理客户端：" + clientInfo + "，线程：" + Thread.currentThread().getName());
        try (InputStream is = socket.getInputStream();
             OutputStream os = socket.getOutputStream();
             BufferedReader reader = new BufferedReader(new InputStreamReader(is));
             PrintWriter writer = new PrintWriter(os, true)) {
            // 读取客户端请求
            String request;
            while ((request = reader.readLine()) != null) {
                if ("exit".equalsIgnoreCase(request)) {
                    writer.println("再见!");
                    break;
                }
                // 模拟业务处理
                String response = processRequest(request);
                writer.println(response);
            }
        } catch (IOException e) {
            System.err.println("处理客户端错误：" + e.getMessage());
        } finally {
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println("客户端连接关闭：" + clientInfo);
        }
    }

    private String processRequest(String request) {
        // 模拟不同的处理时间
        try {
            int processTime = (int)(Math.random() * 3000);
            Thread.sleep(processTime);
            return "处理结果：" + request + " (耗时：" + processTime + "ms)";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "处理中断";
        }
    }
}

5.2.1 线程池工作流程解析

线程池的核心工作原理可以通过以下流程图清晰地展示：

工作流程详解：

1. 任务提交：使用者提交任务到线程池
2. 核心线程判断：检查核心线程池是否已满
   • 未满：创建新线程执行任务
   • 已满：进入下一步
3. 队列判断：检查工作队列是否已满
   • 未满：将任务存储到队列中等待执行
   • 已满：进入下一步
4. 最大线程判断：检查线程池是否达到最大线程数
   • 未满：创建新线程执行任务
   • 已满：执行拒绝策略

这种分层处理机制确保了线程池的高效运行，既避免了过度创建线程，又保证了任务能够得到及时处理。

5.3 高级线程池配置策略

public class AdvancedThreadPoolConfig {
    /**
     * 动态调整线程池参数
     */
    public static class DynamicThreadPool {
        private ThreadPoolExecutor executor;
        private ScheduledExecutorService adjustExecutor;

        public DynamicThreadPool() {
            // 初始配置
            executor = new ThreadPoolExecutor(
                10, 50, 60, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
            );
            // 动态调整线程池参数
            adjustExecutor = Executors.newSingleThreadScheduledExecutor();
            adjustExecutor.scheduleAtFixedRate(this::adjustPoolSize, 0, 10, TimeUnit.SECONDS);
        }

        private void adjustPoolSize() {
            int queueSize = executor.getQueue().size();
            int activeCount = executor.getActiveCount();
            int corePoolSize = executor.getCorePoolSize();
            // 根据队列长度和活动线程数动态调整
            if (queueSize > 800 && activeCount == executor.getMaximumPoolSize()) {
                // 队列满了，且线程数已达上限，可以记录日志或报警
                System.out.println("警告：线程池队列即将满！");
            } else if (queueSize < 100 && activeCount < corePoolSize / 2) {
                // 负载较低，可以适当缩小核心线程数
                int newCoreSize = Math.max(5, corePoolSize - 2);
                executor.setCorePoolSize(newCoreSize);
                System.out.println("降低核心线程数到：" + newCoreSize);
            }
        }

        public void execute(Runnable task) {
            executor.execute(task);
        }

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

    /**
     * 带任务优先级和超时控制的线程池
     */
    public static class PriorityThreadPool {
        private ExecutorService executor;

        public PriorityThreadPool() {
            executor = new ThreadPoolExecutor(
                4, 8, 60, TimeUnit.SECONDS,
                new PriorityBlockingQueue<>(100, Comparator.comparingInt(PriorityTask::getPriority))
            );
        }

        public Future<?> submit(PriorityTask task) {
            return executor.submit(task);
        }

        public void executeWithTimeout(Runnable task, long timeout, TimeUnit unit) {
            Future<?> future = executor.submit(task);
            try {
                future.get(timeout, unit);
            } catch (TimeoutException e) {
                future.cancel(true);
                System.err.println("任务执行超时，已取消");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    static class PriorityTask implements Runnable, Comparable<PriorityTask> {
        private Runnable task;
        private int priority;

        public PriorityTask(Runnable task, int priority) {
            this.task = task;
            this.priority = priority;
        }

        public int getPriority() {
            return priority;
        }

        @Override
        public void run() {
            task.run();
        }

        @Override
        public int compareTo(PriorityTask other) {
            return Integer.compare(other.priority, this.priority); // 降序
        }
    }
}

六、总结

1. Socket 基础：理解 Socket 的基本概念和通信机制是网络编程的基础
2. 多线程必要性：为支持并发连接，必须使用多线程处理 Socket
3. 传统多线程的局限：线程创建销毁开销大、管理复杂、稳定性差
4. 线程池的优势：资源复用、提高响应速度、更好的可管理性