Java多主机通信性能优化的方法

import java.io.*;
import java.net.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.zip.GZIPOutputStream;
/**
 * 优化后的多主机通信示例
 * 核心优化：连接池、精细化线程池、超时配置、数据压缩
 */
public class OptimizedMultiHostCommunication {
    // 1. 精细化配置的线程池（IO密集型）
    private static final ExecutorService executor = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors() * 4, // 核心线程数（IO密集型调高）
            Runtime.getRuntime().availableProcessors() * 8, // 最大线程数
            60L, TimeUnit.SECONDS, // 空闲线程存活时间
            new LinkedBlockingQueue<>(1000), // 任务队列（限制大小，避免OOM）
            new ThreadFactory() { // 自定义线程名，便于排查问题
                private int count = 0;
                @Override
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r);
                    t.setName("host-communication-thread-" + (++count));
                    t.setDaemon(true); // 守护线程，不阻塞程序退出
                    return t;
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：主线程兜底执行，避免任务丢失
    );
    // 2. 连接池（Key：主机IP+端口，Value：可复用的HostClient）
    private static final Map<String, HostClient> connectionPool = new ConcurrentHashMap<>();
    // 连接池最大空闲时间（30秒）
    private static final long MAX_IDLE_TIME = 30_000L;
    public static void main(String[] args) {
        // 定义多主机列表
        List<HostInfo> hostList = new ArrayList<>();
        hostList.add(new HostInfo("192.168.1.100", 8080, "Hello Host 1!"));
        hostList.add(new HostInfo("192.168.1.101", 8080, "Hello Host 2!"));
        hostList.add(new HostInfo("192.168.1.102", 8080, "Hello Host 3!"));
        // 并发处理多主机请求
        hostList.forEach(host -> executor.submit(() -> {
            HostClient client = null;
            try {
                // 从连接池获取/创建连接（复用连接）
                client = getOrCreateClient(host.getIp(), host.getPort());
                // 设置超时（避免阻塞）
                client.setSoTimeout(5000); // 5秒读取超时
                // 发送压缩数据（减少网络传输）
                String compressedMsg = compressMsg(host.getMsg());
                client.sendMsg(compressedMsg);
                // 接收响应
                String response = client.receiveMsg();
                System.out.printf("主机 %s:%d 响应：%s%n", host.getIp(), host.getPort(), response);
            } catch (IOException e) {
                // 失败时移除失效连接，避免复用
                removeInvalidClient(host.getIp(), host.getPort());
                System.err.printf("连接主机 %s:%d 失败：%s%n", host.getIp(), host.getPort(), e.getMessage());
            }
        }));
        // 优雅关闭线程池和连接池
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            executor.shutdown();
            try {
                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                    executor.shutdownNow();
                }
            } catch (InterruptedException e) {
                executor.shutdownNow();
            }
            // 关闭所有连接池中的连接
            connectionPool.values().forEach(client -> {
                try {
                    client.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
            connectionPool.clear();
        }));
    }
    // 从连接池获取客户端（复用连接）
    private static HostClient getOrCreateClient(String ip, int port) throws IOException {
        String key = ip + ":" + port;
        // 双重检查锁，避免并发创建
        if (!connectionPool.containsKey(key)) {
            synchronized (connectionPool) {
                if (!connectionPool.containsKey(key)) {
                    HostClient client = new HostClient(ip, port);
                    connectionPool.put(key, client);
                    // 启动定时清理空闲连接的任务（仅首次创建时启动）
                    if (connectionPool.size() == 1) {
                        scheduleIdleConnectionCleanup();
                    }
                    return client;
                }
            }
        }
        HostClient client = connectionPool.get(key);
        // 检查连接是否有效，无效则重建
        if (client.isClosed() || System.currentTimeMillis() - client.getLastUsedTime() > MAX_IDLE_TIME) {
            client.close();
            HostClient newClient = new HostClient(ip, port);
            connectionPool.put(key, newClient);
            return newClient;
        }
        client.updateLastUsedTime(); // 更新最后使用时间
        return client;
    }
    // 定时清理空闲连接（每10秒执行一次）
    private static void scheduleIdleConnectionCleanup() {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> {
            long now = System.currentTimeMillis();
            connectionPool.entrySet().removeIf(entry -> {
                HostClient client = entry.getValue();
                if (now - client.getLastUsedTime() > MAX_IDLE_TIME) {
                    try {
                        client.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    return true;
                }
                return false;
            });
            // 无连接时关闭调度器
            if (connectionPool.isEmpty()) {
                scheduler.shutdown();
            }
        }, 10, 10, TimeUnit.SECONDS);
    }
    // 移除失效连接
    private static void removeInvalidClient(String ip, int port) {
        String key = ip + ":" + port;
        HostClient client = connectionPool.remove(key);
        if (client != null) {
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    // 压缩消息（减少网络传输量）
    private static String compressMsg(String msg) throws IOException {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        try (GZIPOutputStream gzip = new GZIPOutputStream(bos)) {
            gzip.write(msg.getBytes(StandardCharsets.UTF_8));
        }
        return Base64.getEncoder().encodeToString(bos.toByteArray());
    }
    /**
     * 优化后的主机信息类
     */
    static class HostInfo {
        private final String ip;
        private final int port;
        private final String msg;
        public HostInfo(String ip, int port, String msg) {
            this.ip = ip;
            this.port = port;
            this.msg = msg;
        }
        public String getIp() { return ip; }
        public int getPort() { return port; }
        public String getMsg() { return msg; }
    }
    /**
     * 优化后的主机客户端（支持连接复用、超时、空闲时间跟踪）
     */
    static class HostClient implements AutoCloseable {
        private final Socket socket;
        private final OutputStreamWriter writer;
        private final BufferedReader reader;
        private long lastUsedTime; // 最后使用时间（用于空闲清理）
        public HostClient(String ip, int port) throws IOException {
            // 连接超时配置（3秒）
            this.socket = new Socket();
            this.socket.connect(new InetSocketAddress(ip, port), 3000);
            this.writer = new OutputStreamWriter(socket.getOutputStream(), StandardCharsets.UTF_8);
            this.reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
            this.lastUsedTime = System.currentTimeMillis();
        }
        // 设置读取超时
        public void setSoTimeout(int timeout) throws SocketException {
            socket.setSoTimeout(timeout);
        }
        // 发送消息
        public void sendMsg(String msg) throws IOException {
            writer.write(msg + "\n");
            writer.flush();
            updateLastUsedTime();
        }
        // 接收消息
        public String receiveMsg() throws IOException {
            updateLastUsedTime();
            return reader.readLine();
        }
        // 检查连接是否关闭
        public boolean isClosed() {
            return socket.isClosed() || !socket.isConnected();
        }
        // 更新最后使用时间
        public void updateLastUsedTime() {
            this.lastUsedTime = System.currentTimeMillis();
        }
        // 获取最后使用时间
        public long getLastUsedTime() {
            return lastUsedTime;
        }
        // 关闭连接
        @Override
        public void close() throws IOException {
            if (reader != null) reader.close();
            if (writer != null) writer.close();
            if (socket != null && !socket.isClosed()) socket.close();
        }
    }
}

<!-- Maven 依赖 -->
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.100.Final</version>
</dependency>

// 异步发送请求，聚合结果
List<CompletableFuture<String>> futures = hostList.stream()
        .map(host -> CompletableFuture.supplyAsync(() -> {
            try (HostClient client = new HostClient(host.getIp(), host.getPort())) {
                client.sendMsg(host.getMsg());
                return client.receiveMsg();
            } catch (IOException e) {
                return "失败：" + e.getMessage();
            }
        }, executor))
        .toList();
// 等待所有异步任务完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
// 处理结果
for (int i = 0; i < hostList.size(); i++) {
    System.out.printf("主机 %s:%d 结果：%s%n", 
            hostList.get(i).getIp(), hostList.get(i).getPort(), 
            futures.get(i).join());
}