Java 应用中高效并发执行大量 Linux 命令的策略与实践

import java.io.IOException;

public class BasicCommandExecution {
    public static void main(String[] args) {
        try {
            ProcessBuilder pb = new ProcessBuilder("ls", "-l");
            // 将子进程的标准错误流重定向到标准输出流，方便统一处理
            pb.redirectErrorStream(true);
            Process process = pb.start(); // 启动进程

            // 等待进程执行完成并获取退出码
            int exitCode = process.waitFor();
            System.out.println("Command 'ls -l' exited with code: " + exitCode);

            // 读取命令输出（重要：必须消费输出流，否则可能导致子进程阻塞）
            try (var reader = new java.io.BufferedReader(new java.io.InputStreamReader(process.getInputStream()))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    System.out.println(line);
                }
            }
        } catch (IOException | InterruptedException e) {
            System.err.println("Error executing command: " + e.getMessage());
            Thread.currentThread().interrupt();
        }
    }
}

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.io.IOException;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentCommandLauncher {

    // 存储活跃进程，以便后续管理（如停止）
    private static final Map<Integer, Process> activeProcesses = new ConcurrentHashMap<>();
    // 控制同时启动命令的并发度，避免瞬间创建过多进程
    private static final int MAX_CONCURRENT_LAUNCHES = 200;

    public static void main(String[] args) {
        ExecutorService launchExecutor = Executors.newFixedThreadPool(MAX_CONCURRENT_LAUNCHES);
        int numberOfCommandsToLaunch = 5000; // 假设需要启动5000个socat实例

        System.out.println("开始启动 " + numberOfCommandsToLaunch + " 个 socat 命令...");

        for (int i = 0; i < numberOfCommandsToLaunch; i++) {
            final int commandId = i;
            launchExecutor.submit(() -> {
                // 示例 socat 命令：监听端口 8000+id，转发到 127.0.0.1:80
                String[] command = {"socat", "TCP-LISTEN:" + (8000 + commandId) + ",fork", "TCP:127.0.0.1:80"};
                try {
                    ProcessBuilder pb = new ProcessBuilder(command);
                    pb.redirectErrorStream(true); // 将标准错误流合并到标准输出流

                    Process process = pb.start();
                    activeProcesses.put(commandId, process); // 存储进程对象，以便后续管理

                    // 关键：在新线程中消费进程的输出流，防止阻塞
                    new Thread(() -> {
                        try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
                            String line;
                            while ((line = reader.readLine()) != null) {
                                // 在这里处理或记录socat的输出。
                                // 对于socat这类转发工具，通常只有错误或调试信息才会输出。
                                // System.out.println("Socat " + commandId + " Output: " + line);
                            }
                        } catch (IOException e) {
                            System.err.println("读取 socat " + commandId + " 输出时发生错误: " + e.getMessage());
                        }
                    }).start();

                    // 对于长期运行的socat进程，通常不需要立即调用 process.waitFor()
                    // 如果调用，它会阻塞直到socat进程终止。
                    // 实际应用中，你可能需要一个独立的管理服务来监控和销毁这些进程。

                } catch (IOException e) {
                    System.err.println("启动 socat " + commandId + " 失败: " + e.getMessage());
                }
            });
        }

        launchExecutor.shutdown(); // 关闭线程池，不再接受新任务
        try {
            // 等待所有启动任务完成
            if (!launchExecutor.awaitTermination(10, TimeUnit.MINUTES)) {
                System.err.println("部分 socat 启动任务未能及时完成。");
            }
        } catch (InterruptedException e) {
            System.err.println("启动执行器终止被中断: " + e.getMessage());
            Thread.currentThread().interrupt();
        }

        System.out.println("已成功启动 " + activeProcesses.size() + " 个 socat 命令。现在开始管理它们...");

        // 示例：模拟外部信号服务销毁部分进程
        try {
            Thread.sleep(5000); // 让进程运行一段时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        System.out.println("销毁部分活跃的 socat 进程...");
        int countDestroyed = 0;
        // 销毁前5个启动的进程作为示例
        for (int i = 0; i < 5; i++) {
            Process p = activeProcesses.remove(i);
            if (p != null) {
                p.destroy(); // 发送 SIGTERM 信号
                try {
                    if (!p.waitFor(5, TimeUnit.SECONDS)) { // 等待进程优雅终止
                        p.destroyForcibly(); // 如果未终止，则发送 SIGKILL 信号强制终止
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                countDestroyed++;
            }
        }
        System.out.println("已销毁 " + countDestroyed + " 个进程。");

        // 实际应用中，应有更完善的机制来管理所有剩余进程的生命周期
    }
}