Java Socket 网络编程涉及不同主机间进程通信的端点实现，是开发即时通讯、在线游戏及分布式系统的基础。文章详细阐述了 Socket 的定义、TCP 通信机制及三次握手四次挥手流程，分析了单线程处理连接的局限性及传统多线程模型在资源消耗、管理复杂度和稳定性方面的缺陷。重点介绍了基于线程池的优化方案，包括核心线程复用、任务队列管理、自定义线程工厂与拒绝策略，以及动态调整参数和优先级控制的高级配置，为构建高并发稳定的网络服务提供完整技术参考。
在网络编程中，Socket 是不同主机间进程通信的端点，是实现网络通信的基石。无论是开发即时通讯软件、在线游戏还是分布式系统，都离不开 Socket 技术的支持。本文将深入探讨 Java 中 Socket 的定义、通讯机制，并分析传统多线程模型的缺陷，最后介绍如何使用线程池进行优化。
Socket（套接字）是网络通信的抽象端点，是支持 TCP/IP 协议的网络通信的基本操作单元。在 Java 中，Socket 和 ServerSocket 类分别用于客户端和服务器端的网络通信。
// 服务器端创建 ServerSocket
ServerSocket serverSocket ();
(, );
为了更好地理解 Socket 在网络通信中的作用，我们需要了解它在 OSI 七层模型或 TCP/IP 四层模型中的位置：
如图所示：
Socket 作为会话层的实现，为上层应用提供了统一的网络编程接口，屏蔽了下层复杂的网络细节。
// 服务器端示例
public class SimpleServer {
public static void main(String[] args) throws IOException {
// 1. 创建 ServerSocket，监听指定端口
ServerSocket serverSocket = new ServerSocket(8888);
System.out.println("服务器启动，等待客户端连接...");
// 2. 接受客户端连接
Socket socket = serverSocket.accept();
// 3. 获取输入输出流
InputStream is = socket.getInputStream();
OutputStream os = socket.getOutputStream();
// 4. 数据交换
byte[] buffer = new byte[1024];
int len = is.read(buffer);
String message = new String(buffer, 0, len);
System.out.println("收到客户端消息：" + message);
// 5. 响应客户端
os.write("Hello Client!".getBytes());
// 6. 关闭资源
socket.close();
serverSocket.close();
}
}
// 客户端示例
public class SimpleClient {
public static void main(String[] args) throws IOException {
// 1. 创建 Socket 连接服务器
Socket socket = new Socket("localhost", 8888);
// 2. 获取输入输出流
OutputStream os = socket.getOutputStream();
InputStream is = socket.getInputStream();
// 3. 发送数据
os.write("Hello Server!".getBytes());
os.flush();
// 4. 接收响应
byte[] buffer = new byte[1024];
int len = is.read(buffer);
String response = new String(buffer, 0, len);
System.out.println("服务器响应：" + response);
// 5. 关闭连接
socket.close();
}
}
在建立 TCP 连接时，需要经过三次握手：
断开连接时的四次挥手：
单线程服务器一次只能处理一个客户端连接，这在实际应用中是不可接受的。为了支持并发连接，我们需要引入多线程机制。
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();
}
}
}
}
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;
}
}
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");
}
}
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();
}
}
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 "处理中断";
}
}
}
线程池的核心工作原理可以通过以下流程图清晰地展示：
工作流程详解：
这种分层处理机制确保了线程池的高效运行，既避免了过度创建线程，又保证了任务能够得到及时处理。
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); // 降序
}
}
}
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
查找任何按下的键的javascript键代码、代码、位置和修饰符。 在线工具，Keycode 信息在线工具，online
JavaScript 字符串转义/反转义；Java 风格 \uXXXX（Native2Ascii）编码与解码。 在线工具，Escape 与 Native 编解码在线工具，online
使用 Prettier 在浏览器内格式化 JavaScript 或 HTML 片段。 在线工具，JavaScript / HTML 格式化在线工具，online
Terser 压缩、变量名混淆，或 javascript-obfuscator 高强度混淆（体积会增大）。 在线工具，JavaScript 压缩与混淆在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online