Socket性能优化与调试

1. Socket性能优化

1.1 缓冲区优化

1.1.1 Socket缓冲区设置

import java.io.*;
import java.net.*;

public class SocketBufferOptimization {
    
    public static void optimizeSocketBuffers(Socket socket) throws SocketException {
        // 设置发送缓冲区大小（默认通常是8KB）
        socket.setSendBufferSize(64 * 1024); // 64KB
        
        // 设置接收缓冲区大小（默认通常是8KB）
        socket.setReceiveBufferSize(64 * 1024); // 64KB
        
        // 获取实际设置的缓冲区大小
        int actualSendBuffer = socket.getSendBufferSize();
        int actualReceiveBuffer = socket.getReceiveBufferSize();
        
        System.out.println("发送缓冲区大小: " + actualSendBuffer + " 字节");
        System.out.println("接收缓冲区大小: " + actualReceiveBuffer + " 字节");
    }
    
    public static void optimizeServerSocket(ServerSocket serverSocket) throws SocketException {
        // 设置接收缓冲区大小
        serverSocket.setReceiveBufferSize(128 * 1024); // 128KB
        
        // 启用地址重用
        serverSocket.setReuseAddress(true);
        
        System.out.println("服务器Socket缓冲区大小: " + serverSocket.getReceiveBufferSize());
        System.out.println("地址重用: " + serverSocket.getReuseAddress());
    }
    
    // 应用层缓冲区优化
    public static class OptimizedBufferedStreams {
        private static final int BUFFER_SIZE = 32 * 1024; // 32KB
        
        public static BufferedInputStream createOptimizedInput(InputStream input) {
            return new BufferedInputStream(input, BUFFER_SIZE);
        }
        
        public static BufferedOutputStream createOptimizedOutput(OutputStream output) {
            return new BufferedOutputStream(output, BUFFER_SIZE);
        }
    }
}

1.1.2 缓冲区大小测试

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

public class BufferSizePerformanceTest {
    private static final String TEST_HOST = "localhost";
    private static final int TEST_PORT = 9999;
    private static final int DATA_SIZE = 10 * 1024 * 1024; // 10MB
    
    public static void main(String[] args) throws Exception {
        // 测试不同缓冲区大小的性能
        int[] bufferSizes = {1024, 4096, 8192, 16384, 32768, 65536};
        
        for (int bufferSize : bufferSizes) {
            System.out.println("\n测试缓冲区大小: " + bufferSize + " 字节");
            testBufferSize(bufferSize);
        }
    }
    
    private static void testBufferSize(int bufferSize) throws Exception {
        // 启动测试服务器
        ExecutorService executor = Executors.newFixedThreadPool(2);
        
        Future<?> serverFuture = executor.submit(() -> {
            try {
                runTestServer(bufferSize);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        
        // 等待服务器启动
        Thread.sleep(100);
        
        // 运行客户端测试
        long startTime = System.currentTimeMillis();
        runTestClient(bufferSize);
        long endTime = System.currentTimeMillis();
        
        System.out.println("传输时间: " + (endTime - startTime) + " ms");
        System.out.println("传输速度: " + (DATA_SIZE / 1024.0 / 1024.0) / ((endTime - startTime) / 1000.0) + " MB/s");
        
        executor.shutdown();
    }
    
    private static void runTestServer(int bufferSize) throws Exception {
        try (ServerSocket serverSocket = new ServerSocket(TEST_PORT)) {
            serverSocket.setReceiveBufferSize(bufferSize);
            
            try (Socket clientSocket = serverSocket.accept()) {
                clientSocket.setReceiveBufferSize(bufferSize);
                
                try (InputStream input = new BufferedInputStream(
                        clientSocket.getInputStream(), bufferSize)) {
                    
                    byte[] buffer = new byte[bufferSize];
                    int totalReceived = 0;
                    int bytesRead;
                    
                    while (totalReceived < DATA_SIZE && 
                           (bytesRead = input.read(buffer)) != -1) {
                        totalReceived += bytesRead;
                    }
                }
            }
        }
    }
    
    private static void runTestClient(int bufferSize) throws Exception {
        try (Socket socket = new Socket(TEST_HOST, TEST_PORT)) {
            socket.setSendBufferSize(bufferSize);
            
            try (OutputStream output = new BufferedOutputStream(
                    socket.getOutputStream(), bufferSize)) {
                
                byte[] data = new byte[bufferSize];
                int totalSent = 0;
                
                while (totalSent < DATA_SIZE) {
                    int toSend = Math.min(bufferSize, DATA_SIZE - totalSent);
                    output.write(data, 0, toSend);
                    totalSent += toSend;
                }
                
                output.flush();
            }
        }
    }
}

1.2 连接池优化

1.2.1 Socket连接池实现

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

public class SocketConnectionPool {
    private final String host;
    private final int port;
    private final int maxConnections;
    private final int connectionTimeout;
    private final int socketTimeout;
    
    private final BlockingQueue<PooledSocket> availableConnections;
    private final AtomicInteger activeConnections = new AtomicInteger(0);
    private volatile boolean closed = false;
    
    public SocketConnectionPool(String host, int port, int maxConnections) {
        this(host, port, maxConnections, 5000, 30000);
    }
    
    public SocketConnectionPool(String host, int port, int maxConnections, 
                              int connectionTimeout, int socketTimeout) {
        this.host = host;
        this.port = port;
        this.maxConnections = maxConnections;
        this.connectionTimeout = connectionTimeout;
        this.socketTimeout = socketTimeout;
        this.availableConnections = new LinkedBlockingQueue<>(maxConnections);
    }
    
    public PooledSocket getConnection() throws IOException, InterruptedException {
        if (closed) {
            throw new IllegalStateException("连接池已关闭");
        }
        
        // 尝试从池中获取可用连接
        PooledSocket pooledSocket = availableConnections.poll();
        
        if (pooledSocket != null && pooledSocket.isValid()) {
            return pooledSocket;
        }
        
        // 如果没有可用连接且未达到最大连接数，创建新连接
        if (activeConnections.get() < maxConnections) {
            Socket socket = createNewSocket();
            pooledSocket = new PooledSocket(socket, this);
            activeConnections.incrementAndGet();
            return pooledSocket;
        }
        
        // 等待可用连接
        pooledSocket = availableConnections.take();
        if (!pooledSocket.isValid()) {
            // 连接无效，创建新连接
            Socket socket = createNewSocket();
            return new PooledSocket(socket, this);
        }
        
        return pooledSocket;
    }
    
    private Socket createNewSocket() throws IOException {
        Socket socket = new Socket();
        socket.connect(new InetSocketAddress(host, port), connectionTimeout);
        socket.setSoTimeout(socketTimeout);
        
        // 优化Socket设置
        socket.setTcpNoDelay(true);
        socket.setKeepAlive(true);
        socket.setSendBufferSize(64 * 1024);
        socket.setReceiveBufferSize(64 * 1024);
        
        return socket;
    }
    
    void returnConnection(PooledSocket pooledSocket) {
        if (closed || !pooledSocket.isValid()) {
            closeSocket(pooledSocket.getSocket());
            activeConnections.decrementAndGet();
            return;
        }
        
        if (!availableConnections.offer(pooledSocket)) {
            // 池已满，关闭连接
            closeSocket(pooledSocket.getSocket());
            activeConnections.decrementAndGet();
        }
    }
    
    private void closeSocket(Socket socket) {
        try {
            if (socket != null && !socket.isClosed()) {
                socket.close();
            }
        } catch (IOException e) {
            // 忽略关闭异常
        }
    }
    
    public void close() {
        closed = true;
        
        // 关闭所有可用连接
        PooledSocket pooledSocket;
        while ((pooledSocket = availableConnections.poll()) != null) {
            closeSocket(pooledSocket.getSocket());
        }
        
        System.out.println("连接池已关闭，活跃连接数: " + activeConnections.get());
    }
    
    public int getActiveConnections() {
        return activeConnections.get();
    }
    
    public int getAvailableConnections() {
        return availableConnections.size();
    }
    
    // 连接池状态监控
    public void printStatus() {
        System.out.println("连接池状态:");
        System.out.println("  最大连接数: " + maxConnections);
        System.out.println("  活跃连接数: " + activeConnections.get());
        System.out.println("  可用连接数: " + availableConnections.size());
        System.out.println("  连接池状态: " + (closed ? "已关闭" : "运行中"));
    }
}

class PooledSocket implements AutoCloseable {
    private final Socket socket;
    private final SocketConnectionPool pool;
    private volatile boolean inUse = true;
    
    public PooledSocket(Socket socket, SocketConnectionPool pool) {
        this.socket = socket;
        this.pool = pool;
    }
    
    public Socket getSocket() {
        return socket;
    }
    
    public InputStream getInputStream() throws IOException {
        return socket.getInputStream();
    }
    
    public OutputStream getOutputStream() throws IOException {
        return socket.getOutputStream();
    }
    
    public boolean isValid() {
        return socket != null && !socket.isClosed() && socket.isConnected();
    }
    
    @Override
    public void close() {
        if (inUse) {
            inUse = false;
            pool.returnConnection(this);
        }
    }
    
    // 强制关闭连接（不返回池中）
    public void forceClose() {
        inUse = false;
        try {
            if (socket != null && !socket.isClosed()) {
                socket.close();
            }
        } catch (IOException e) {
            // 忽略关闭异常
        }
    }
}

1.2.2 连接池使用示例

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

public class ConnectionPoolExample {
    
    public static void main(String[] args) throws Exception {
        // 创建连接池
        SocketConnectionPool pool = new SocketConnectionPool("httpbin.org", 80, 10);
        
        // 创建线程池进行并发测试
        ExecutorService executor = Executors.newFixedThreadPool(20);
        
        // 提交多个任务
        for (int i = 0; i < 50; i++) {
            final int taskId = i;
            executor.submit(() -> {
                try {
                    performHttpRequest(pool, taskId);
                } catch (Exception e) {
                    System.err.println("任务 " + taskId + " 异常: " + e.getMessage());
                }
            });
        }
        
        // 定期打印连接池状态
        ScheduledExecutorService monitor = Executors.newScheduledThreadPool(1);
        monitor.scheduleAtFixedRate(pool::printStatus, 1, 2, TimeUnit.SECONDS);
        
        // 等待所有任务完成
        executor.shutdown();
        executor.awaitTermination(30, TimeUnit.SECONDS);
        
        // 关闭监控和连接池
        monitor.shutdown();
        pool.close();
    }
    
    private static void performHttpRequest(SocketConnectionPool pool, int taskId) 
            throws Exception {
        
        try (PooledSocket pooledSocket = pool.getConnection()) {
            // 发送HTTP请求
            PrintWriter writer = new PrintWriter(pooledSocket.getOutputStream(), true);
            writer.println("GET /get HTTP/1.1");
            writer.println("Host: httpbin.org");
            writer.println("Connection: keep-alive");
            writer.println();
            
            // 读取响应
            BufferedReader reader = new BufferedReader(
                new InputStreamReader(pooledSocket.getInputStream()));
            
            String line;
            int lineCount = 0;
            while ((line = reader.readLine()) != null && lineCount < 20) {
                if (lineCount == 0) {
                    System.out.println("任务 " + taskId + " 响应: " + line);
                }
                lineCount++;
            }
            
            // 连接会自动返回到池中
        }
    }
}

1.3 NIO优化

1.3.1 NIO服务器优化

import java.io.IOException;
import java.net.*;
import java.nio.*;
import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.*;

public class OptimizedNIOServer {
    private static final int PORT = 9090;
    private static final int BUFFER_SIZE = 8192;
    private static final int MAX_CONNECTIONS = 1000;
    
    private Selector selector;
    private ServerSocketChannel serverChannel;
    private volatile boolean running = false;
    
    // 使用多个Selector提高并发性能
    private final int selectorCount = Runtime.getRuntime().availableProcessors();
    private final Selector[] selectors = new Selector[selectorCount];
    private final ExecutorService selectorThreads = Executors.newFixedThreadPool(selectorCount);
    private int currentSelector = 0;
    
    public void start() throws IOException {
        // 初始化主Selector
        selector = Selector.open();
        
        // 初始化工作Selector
        for (int i = 0; i < selectorCount; i++) {
            selectors[i] = Selector.open();
            final int index = i;
            selectorThreads.submit(() -> runWorkerSelector(index));
        }
        
        // 配置服务器通道
        serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        
        // 优化服务器Socket设置
        ServerSocket serverSocket = serverChannel.socket();
        serverSocket.setReuseAddress(true);
        serverSocket.setReceiveBufferSize(64 * 1024);
        
        serverSocket.bind(new InetSocketAddress(PORT));
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        
        running = true;
        System.out.println("优化的NIO服务器启动，端口: " + PORT);
        System.out.println("Selector数量: " + selectorCount);
        
        runMainSelector();
    }
    
    private void runMainSelector() {
        while (running) {
            try {
                int readyChannels = selector.select(1000);
                
                if (readyChannels == 0) {
                    continue;
                }
                
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
                
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    keyIterator.remove();
                    
                    if (!key.isValid()) {
                        continue;
                    }
                    
                    if (key.isAcceptable()) {
                        handleAccept(key);
                    }
                }
                
            } catch (IOException e) {
                System.err.println("主Selector异常: " + e.getMessage());
            }
        }
    }
    
    private void runWorkerSelector(int index) {
        Selector workerSelector = selectors[index];
        
        while (running) {
            try {
                int readyChannels = workerSelector.select(1000);
                
                if (readyChannels == 0) {
                    continue;
                }
                
                Set<SelectionKey> selectedKeys = workerSelector.selectedKeys();
                Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
                
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    keyIterator.remove();
                    
                    if (!key.isValid()) {
                        continue;
                    }
                    
                    if (key.isReadable()) {
                        handleRead(key);
                    } else if (key.isWritable()) {
                        handleWrite(key);
                    }
                }
                
            } catch (IOException e) {
                System.err.println("工作Selector " + index + " 异常: " + e.getMessage());
            }
        }
    }
    
    private void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
        SocketChannel clientChannel = serverChannel.accept();
        
        if (clientChannel != null) {
            // 优化客户端Socket设置
            Socket socket = clientChannel.socket();
            socket.setTcpNoDelay(true);
            socket.setKeepAlive(true);
            socket.setSendBufferSize(32 * 1024);
            socket.setReceiveBufferSize(32 * 1024);
            
            clientChannel.configureBlocking(false);
            
            // 负载均衡到工作Selector
            Selector workerSelector = selectors[currentSelector % selectorCount];
            currentSelector++;
            
            // 唤醒Selector并注册通道
            workerSelector.wakeup();
            clientChannel.register(workerSelector, SelectionKey.OP_READ, 
                                 new ClientContext());
            
            System.out.println("客户端连接: " + clientChannel.getRemoteAddress() + 
                             " -> Selector " + (currentSelector - 1) % selectorCount);
        }
    }
    
    private void handleRead(SelectionKey key) {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        ClientContext context = (ClientContext) key.attachment();
        
        try {
            ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
            int bytesRead = clientChannel.read(buffer);
            
            if (bytesRead > 0) {
                buffer.flip();
                
                // 处理接收到的数据
                byte[] data = new byte[buffer.remaining()];
                buffer.get(data);
                
                String message = new String(data).trim();
                System.out.println("收到消息: " + message);
                
                // 准备响应
                String response = "Echo: " + message + "\n";
                context.setResponse(response);
                
                // 注册写事件
                key.interestOps(SelectionKey.OP_WRITE);
                
            } else if (bytesRead == -1) {
                // 客户端断开连接
                System.out.println("客户端断开: " + clientChannel.getRemoteAddress());
                key.cancel();
                clientChannel.close();
            }
            
        } catch (IOException e) {
            System.err.println("读取数据异常: " + e.getMessage());
            try {
                key.cancel();
                clientChannel.close();
            } catch (IOException ex) {
                // 忽略关闭异常
            }
        }
    }
    
    private void handleWrite(SelectionKey key) {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        ClientContext context = (ClientContext) key.attachment();
        
        try {
            String response = context.getResponse();
            if (response != null) {
                ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
                clientChannel.write(buffer);
                
                if (!buffer.hasRemaining()) {
                    // 写入完成，切换回读模式
                    context.setResponse(null);
                    key.interestOps(SelectionKey.OP_READ);
                }
            }
            
        } catch (IOException e) {
            System.err.println("写入数据异常: " + e.getMessage());
            try {
                key.cancel();
                clientChannel.close();
            } catch (IOException ex) {
                // 忽略关闭异常
            }
        }
    }
    
    public void stop() throws IOException {
        running = false;
        
        if (selector != null) {
            selector.wakeup();
            selector.close();
        }
        
        for (Selector workerSelector : selectors) {
            if (workerSelector != null) {
                workerSelector.wakeup();
                workerSelector.close();
            }
        }
        
        if (serverChannel != null) {
            serverChannel.close();
        }
        
        selectorThreads.shutdown();
        System.out.println("优化的NIO服务器已停止");
    }
    
    private static class ClientContext {
        private String response;
        
        public String getResponse() {
            return response;
        }
        
        public void setResponse(String response) {
            this.response = response;
        }
    }
    
    public static void main(String[] args) {
        OptimizedNIOServer server = new OptimizedNIOServer();
        
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                server.stop();
            } catch (IOException e) {
                System.err.println("停止服务器异常: " + e.getMessage());
            }
        }));
        
        try {
            server.start();
        } catch (IOException e) {
            System.err.println("启动NIO服务器失败: " + e.getMessage());
        }
    }
}

2. Socket调试技术

2.1 网络监控工具

2.1.1 Socket状态监控

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

public class SocketMonitor {
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    private final Map<String, SocketStats> socketStats = new ConcurrentHashMap<>();
    
    public void startMonitoring() {
        scheduler.scheduleAtFixedRate(this::collectStats, 0, 5, TimeUnit.SECONDS);
        scheduler.scheduleAtFixedRate(this::printStats, 10, 10, TimeUnit.SECONDS);
    }
    
    public void stopMonitoring() {
        scheduler.shutdown();
    }
    
    private void collectStats() {
        try {
            // 获取系统网络统计信息
            collectSystemNetworkStats();
            
            // 获取JVM内存使用情况
            collectMemoryStats();
            
            // 获取线程统计信息
            collectThreadStats();
            
        } catch (Exception e) {
            System.err.println("收集统计信息异常: " + e.getMessage());
        }
    }
    
    private void collectSystemNetworkStats() {
        try {
            // 在Windows上使用netstat命令
            Process process = Runtime.getRuntime().exec("netstat -an");
            BufferedReader reader = new BufferedReader(
                new InputStreamReader(process.getInputStream()));
            
            Map<String, Integer> connectionCounts = new HashMap<>();
            String line;
            
            while ((line = reader.readLine()) != null) {
                if (line.contains("TCP") || line.contains("UDP")) {
                    String[] parts = line.trim().split("\\s+");
                    if (parts.length >= 4) {
                        String state = parts.length > 4 ? parts[4] : "UDP";
                        connectionCounts.merge(state, 1, Integer::sum);
                    }
                }
            }
            
            // 更新统计信息
            for (Map.Entry<String, Integer> entry : connectionCounts.entrySet()) {
                String state = entry.getKey();
                int count = entry.getValue();
                
                socketStats.computeIfAbsent("CONNECTION_" + state, 
                    k -> new SocketStats()).updateValue(count);
            }
            
        } catch (IOException e) {
            System.err.println("收集网络统计信息失败: " + e.getMessage());
        }
    }
    
    private void collectMemoryStats() {
        MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
        MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();
        MemoryUsage nonHeapUsage = memoryBean.getNonHeapMemoryUsage();
        
        socketStats.computeIfAbsent("MEMORY_HEAP_USED", 
            k -> new SocketStats()).updateValue(heapUsage.getUsed() / 1024 / 1024);
        socketStats.computeIfAbsent("MEMORY_HEAP_MAX", 
            k -> new SocketStats()).updateValue(heapUsage.getMax() / 1024 / 1024);
        socketStats.computeIfAbsent("MEMORY_NONHEAP_USED", 
            k -> new SocketStats()).updateValue(nonHeapUsage.getUsed() / 1024 / 1024);
    }
    
    private void collectThreadStats() {
        ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
        
        socketStats.computeIfAbsent("THREAD_COUNT", 
            k -> new SocketStats()).updateValue(threadBean.getThreadCount());
        socketStats.computeIfAbsent("THREAD_PEAK", 
            k -> new SocketStats()).updateValue(threadBean.getPeakThreadCount());
        socketStats.computeIfAbsent("THREAD_DAEMON", 
            k -> new SocketStats()).updateValue(threadBean.getDaemonThreadCount());
    }
    
    private void printStats() {
        System.out.println("\n=== Socket监控统计 ===" + new Date());
        
        // 网络连接统计
        System.out.println("\n网络连接状态:");
        socketStats.entrySet().stream()
            .filter(entry -> entry.getKey().startsWith("CONNECTION_"))
            .forEach(entry -> {
                String state = entry.getKey().substring(11);
                SocketStats stats = entry.getValue();
                System.out.printf("  %s: 当前=%d, 平均=%.1f, 最大=%d\n", 
                    state, stats.getCurrentValue(), stats.getAverageValue(), stats.getMaxValue());
            });
        
        // 内存使用统计
        System.out.println("\n内存使用情况 (MB):");
        socketStats.entrySet().stream()
            .filter(entry -> entry.getKey().startsWith("MEMORY_"))
            .forEach(entry -> {
                String type = entry.getKey().substring(7);
                SocketStats stats = entry.getValue();
                System.out.printf("  %s: 当前=%d, 平均=%.1f, 最大=%d\n", 
                    type, stats.getCurrentValue(), stats.getAverageValue(), stats.getMaxValue());
            });
        
        // 线程统计
        System.out.println("\n线程统计:");
        socketStats.entrySet().stream()
            .filter(entry -> entry.getKey().startsWith("THREAD_"))
            .forEach(entry -> {
                String type = entry.getKey().substring(7);
                SocketStats stats = entry.getValue();
                System.out.printf("  %s: 当前=%d, 平均=%.1f, 最大=%d\n", 
                    type, stats.getCurrentValue(), stats.getAverageValue(), stats.getMaxValue());
            });
    }
    
    private static class SocketStats {
        private long currentValue;
        private long maxValue;
        private long totalValue;
        private int sampleCount;
        
        public synchronized void updateValue(long value) {
            this.currentValue = value;
            this.maxValue = Math.max(maxValue, value);
            this.totalValue += value;
            this.sampleCount++;
        }
        
        public synchronized long getCurrentValue() {
            return currentValue;
        }
        
        public synchronized long getMaxValue() {
            return maxValue;
        }
        
        public synchronized double getAverageValue() {
            return sampleCount > 0 ? (double) totalValue / sampleCount : 0;
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        SocketMonitor monitor = new SocketMonitor();
        monitor.startMonitoring();
        
        System.out.println("Socket监控已启动，按Ctrl+C停止...");
        
        Runtime.getRuntime().addShutdownHook(new Thread(monitor::stopMonitoring));
        
        // 保持程序运行
        Thread.currentThread().join();
    }
}

2.2 网络包分析

2.2.1 简单包捕获工具

import java.io.*;
import java.net.*;
import java.nio.*;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;

public class SimplePacketCapture {
    private static final SimpleDateFormat DATE_FORMAT = 
        new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    
    private final String targetHost;
    private final int targetPort;
    private final PrintWriter logWriter;
    private volatile boolean capturing = false;
    
    public SimplePacketCapture(String targetHost, int targetPort, String logFile) 
            throws IOException {
        this.targetHost = targetHost;
        this.targetPort = targetPort;
        this.logWriter = new PrintWriter(new FileWriter(logFile, true));
    }
    
    public void startCapture() {
        capturing = true;
        System.out.println("开始捕获网络包: " + targetHost + ":" + targetPort);
        
        try (ServerSocket proxyServer = new ServerSocket(targetPort + 1000)) {
            System.out.println("代理服务器启动在端口: " + (targetPort + 1000));
            
            while (capturing) {
                try {
                    Socket clientSocket = proxyServer.accept();
                    new Thread(() -> handleConnection(clientSocket)).start();
                } catch (IOException e) {
                    if (capturing) {
                        System.err.println("接受连接异常: " + e.getMessage());
                    }
                }
            }
        } catch (IOException e) {
            System.err.println("启动代理服务器失败: " + e.getMessage());
        }
    }
    
    private void handleConnection(Socket clientSocket) {
        String clientAddress = clientSocket.getRemoteSocketAddress().toString();
        logPacket("CONNECTION", "客户端连接: " + clientAddress, null);
        
        try (Socket targetSocket = new Socket(targetHost, targetPort)) {
            logPacket("CONNECTION", "连接到目标服务器: " + targetHost + ":" + targetPort, null);
            
            // 启动双向数据转发
            Thread clientToTarget = new Thread(() -> 
                forwardData(clientSocket, targetSocket, "CLIENT->SERVER"));
            Thread targetToClient = new Thread(() -> 
                forwardData(targetSocket, clientSocket, "SERVER->CLIENT"));
            
            clientToTarget.start();
            targetToClient.start();
            
            clientToTarget.join();
            targetToClient.join();
            
        } catch (Exception e) {
            logPacket("ERROR", "连接处理异常: " + e.getMessage(), null);
        } finally {
            try {
                clientSocket.close();
                logPacket("CONNECTION", "客户端连接关闭: " + clientAddress, null);
            } catch (IOException e) {
                // 忽略关闭异常
            }
        }
    }
    
    private void forwardData(Socket from, Socket to, String direction) {
        try (InputStream input = from.getInputStream();
             OutputStream output = to.getOutputStream()) {
            
            byte[] buffer = new byte[4096];
            int bytesRead;
            
            while ((bytesRead = input.read(buffer)) != -1) {
                output.write(buffer, 0, bytesRead);
                output.flush();
                
                // 记录数据包
                byte[] data = Arrays.copyOf(buffer, bytesRead);
                logPacket("DATA", direction + " (" + bytesRead + " 字节)", data);
            }
            
        } catch (IOException e) {
            logPacket("ERROR", direction + " 数据转发异常: " + e.getMessage(), null);
        }
    }
    
    private synchronized void logPacket(String type, String description, byte[] data) {
        String timestamp = DATE_FORMAT.format(new Date());
        
        logWriter.println("[" + timestamp + "] [" + type + "] " + description);
        
        if (data != null && data.length > 0) {
            // 记录十六进制数据
            logWriter.println("  HEX: " + bytesToHex(data));
            
            // 记录可打印字符
            String text = bytesToText(data);
            if (!text.isEmpty()) {
                logWriter.println("  TEXT: " + text);
            }
        }
        
        logWriter.println();
        logWriter.flush();
        
        // 同时输出到控制台
        System.out.println("[" + timestamp + "] [" + type + "] " + description);
    }
    
    private String bytesToHex(byte[] bytes) {
        StringBuilder hex = new StringBuilder();
        for (int i = 0; i < Math.min(bytes.length, 256); i++) {
            if (i > 0 && i % 16 == 0) {
                hex.append("\n       ");
            }
            hex.append(String.format("%02X ", bytes[i]));
        }
        if (bytes.length > 256) {
            hex.append("... (截断)");
        }
        return hex.toString();
    }
    
    private String bytesToText(byte[] bytes) {
        StringBuilder text = new StringBuilder();
        for (int i = 0; i < Math.min(bytes.length, 256); i++) {
            char c = (char) bytes[i];
            if (c >= 32 && c <= 126) {
                text.append(c);
            } else {
                text.append('.');
            }
        }
        return text.toString().trim();
    }
    
    public void stopCapture() {
        capturing = false;
        logWriter.close();
        System.out.println("网络包捕获已停止");
    }
    
    public static void main(String[] args) throws IOException {
        if (args.length < 2) {
            System.out.println("用法: java SimplePacketCapture <目标主机> <目标端口> [日志文件]");
            System.out.println("示例: java SimplePacketCapture www.baidu.com 80 capture.log");
            return;
        }
        
        String targetHost = args[0];
        int targetPort = Integer.parseInt(args[1]);
        String logFile = args.length > 2 ? args[2] : "packet_capture.log";
        
        SimplePacketCapture capture = new SimplePacketCapture(targetHost, targetPort, logFile);
        
        Runtime.getRuntime().addShutdownHook(new Thread(capture::stopCapture));
        
        capture.startCapture();
    }
}

2.3 性能分析工具

2.3.1 Socket性能测试工具

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

public class SocketPerformanceTester {
    
    public static class PerformanceResult {
        private final long totalRequests;
        private final long successfulRequests;
        private final long failedRequests;
        private final long totalTime;
        private final long minTime;
        private final long maxTime;
        private final double averageTime;
        private final double throughput;
        
        public PerformanceResult(long totalRequests, long successfulRequests, 
                               long failedRequests, long totalTime, 
                               long minTime, long maxTime, double averageTime) {
            this.totalRequests = totalRequests;
            this.successfulRequests = successfulRequests;
            this.failedRequests = failedRequests;
            this.totalTime = totalTime;
            this.minTime = minTime;
            this.maxTime = maxTime;
            this.averageTime = averageTime;
            this.throughput = successfulRequests > 0 ? 
                (double) successfulRequests / (totalTime / 1000.0) : 0;
        }
        
        public void printResults() {
            System.out.println("\n=== 性能测试结果 ===");
            System.out.println("总请求数: " + totalRequests);
            System.out.println("成功请求数: " + successfulRequests);
            System.out.println("失败请求数: " + failedRequests);
            System.out.println("成功率: " + String.format("%.2f%%", 
                (double) successfulRequests / totalRequests * 100));
            System.out.println("总耗时: " + totalTime + " ms");
            System.out.println("最小响应时间: " + minTime + " ms");
            System.out.println("最大响应时间: " + maxTime + " ms");
            System.out.println("平均响应时间: " + String.format("%.2f ms", averageTime));
            System.out.println("吞吐量: " + String.format("%.2f 请求/秒", throughput));
        }
    }
    
    public static PerformanceResult testSocketPerformance(
            String host, int port, int concurrency, int totalRequests, 
            String testData) {
        
        System.out.println("开始Socket性能测试...");
        System.out.println("目标: " + host + ":" + port);
        System.out.println("并发数: " + concurrency);
        System.out.println("总请求数: " + totalRequests);
        
        ExecutorService executor = Executors.newFixedThreadPool(concurrency);
        CountDownLatch latch = new CountDownLatch(totalRequests);
        
        AtomicLong successCount = new AtomicLong(0);
        AtomicLong failCount = new AtomicLong(0);
        AtomicLong minTime = new AtomicLong(Long.MAX_VALUE);
        AtomicLong maxTime = new AtomicLong(0);
        AtomicLong totalTime = new AtomicLong(0);
        
        long startTime = System.currentTimeMillis();
        
        // 提交测试任务
        for (int i = 0; i < totalRequests; i++) {
            final int requestId = i;
            executor.submit(() -> {
                try {
                    long requestStart = System.currentTimeMillis();
                    
                    boolean success = performSingleRequest(host, port, testData, requestId);
                    
                    long requestTime = System.currentTimeMillis() - requestStart;
                    
                    if (success) {
                        successCount.incrementAndGet();
                        
                        // 更新时间统计
                        minTime.updateAndGet(current -> Math.min(current, requestTime));
                        maxTime.updateAndGet(current -> Math.max(current, requestTime));
                        totalTime.addAndGet(requestTime);
                    } else {
                        failCount.incrementAndGet();
                    }
                    
                } catch (Exception e) {
                    failCount.incrementAndGet();
                    System.err.println("请求 " + requestId + " 异常: " + e.getMessage());
                } finally {
                    latch.countDown();
                }
            });
        }
        
        try {
            // 等待所有请求完成
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        long endTime = System.currentTimeMillis();
        long totalTestTime = endTime - startTime;
        
        executor.shutdown();
        
        // 计算平均响应时间
        double averageTime = successCount.get() > 0 ? 
            (double) totalTime.get() / successCount.get() : 0;
        
        return new PerformanceResult(
            totalRequests, successCount.get(), failCount.get(), 
            totalTestTime, minTime.get(), maxTime.get(), averageTime
        );
    }
    
    private static boolean performSingleRequest(String host, int port, 
                                              String testData, int requestId) {
        try (Socket socket = new Socket()) {
            // 设置连接超时
            socket.connect(new InetSocketAddress(host, port), 5000);
            socket.setSoTimeout(10000);
            
            // 发送测试数据
            try (PrintWriter writer = new PrintWriter(socket.getOutputStream(), true);
                 BufferedReader reader = new BufferedReader(
                     new InputStreamReader(socket.getInputStream()))) {
                
                writer.println(testData + " (请求ID: " + requestId + ")");
                
                // 读取响应
                String response = reader.readLine();
                
                return response != null && !response.isEmpty();
            }
            
        } catch (IOException e) {
            return false;
        }
    }
    
    // 连接池性能测试
    public static void testConnectionPoolPerformance() {
        System.out.println("\n=== 连接池性能测试 ===");
        
        // 测试不同连接池大小的性能
        int[] poolSizes = {1, 5, 10, 20, 50};
        int concurrency = 20;
        int requestsPerTest = 100;
        
        for (int poolSize : poolSizes) {
            System.out.println("\n测试连接池大小: " + poolSize);
            
            long startTime = System.currentTimeMillis();
            
            // 模拟连接池测试
            ExecutorService executor = Executors.newFixedThreadPool(concurrency);
            CountDownLatch latch = new CountDownLatch(requestsPerTest);
            Semaphore connectionPool = new Semaphore(poolSize);
            
            for (int i = 0; i < requestsPerTest; i++) {
                executor.submit(() -> {
                    try {
                        // 获取连接
                        connectionPool.acquire();
                        
                        // 模拟请求处理时间
                        Thread.sleep(10 + (int)(Math.random() * 20));
                        
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    } finally {
                        // 释放连接
                        connectionPool.release();
                        latch.countDown();
                    }
                });
            }
            
            try {
                latch.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            
            long endTime = System.currentTimeMillis();
            
            System.out.println("  完成时间: " + (endTime - startTime) + " ms");
            System.out.println("  平均吞吐量: " + 
                String.format("%.2f 请求/秒", 
                    (double) requestsPerTest / ((endTime - startTime) / 1000.0)));
            
            executor.shutdown();
        }
    }
    
    public static void main(String[] args) {
        if (args.length < 2) {
            System.out.println("用法: java SocketPerformanceTester <主机> <端口> [并发数] [请求数]");
            System.out.println("示例: java SocketPerformanceTester localhost 8080 10 100");
            return;
        }
        
        String host = args[0];
        int port = Integer.parseInt(args[1]);
        int concurrency = args.length > 2 ? Integer.parseInt(args[2]) : 10;
        int totalRequests = args.length > 3 ? Integer.parseInt(args[3]) : 100;
        
        String testData = "Hello, this is a performance test message!";
        
        // 执行性能测试
        PerformanceResult result = testSocketPerformance(
            host, port, concurrency, totalRequests, testData);
        
        result.printResults();
        
        // 执行连接池性能测试
        testConnectionPoolPerformance();
    }
}

3. 总结

3.1 性能优化要点

1. 缓冲区优化：

   • 合理设置Socket缓冲区大小
   • 使用应用层缓冲提高I/O效率
   • 根据网络环境调整缓冲区参数

2. 连接管理：

   • 使用连接池减少连接开销
   • 合理设置连接超时参数
   • 及时释放无用连接

3. 并发优化：

   • 使用NIO提高并发处理能力
   • 多Selector架构分散负载
   • 线程池管理减少线程创建开销

3.2 调试技术要点

1. 监控工具：

   • 实时监控网络连接状态
   • 跟踪内存和线程使用情况
   • 记录关键性能指标

2. 包分析：

   • 捕获和分析网络数据包
   • 识别网络通信问题
   • 验证协议实现正确性

3. 性能测试：

   • 压力测试验证系统容量
   • 基准测试对比不同实现
   • 持续监控生产环境性能

3.3 最佳实践

1. 设计阶段：

   • 选择合适的I/O模型
   • 设计高效的协议格式
   • 考虑扩展性和可维护性

2. 实现阶段：

   • 遵循性能优化原则
   • 添加完善的监控和日志
   • 实现优雅的错误处理

3. 测试阶段：

   • 进行充分的性能测试
   • 模拟各种网络环境
   • 验证异常情况处理

4. 运维阶段：

   • 持续监控系统性能
   • 定期分析性能瓶颈
   • 根据实际情况调优参数

通过系统的性能优化和调试技术，可以构建高性能、稳定可靠的Socket应用程序。