在Java中高效管理与执行大量Linux命令

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.IOException;

public class CommandExecutor {

    public static void executeCommand(String command) {
        ProcessBuilder processBuilder = new ProcessBuilder();
        processBuilder.command("bash", "-c", command); // 使用bash -c 执行命令

        try {
            Process process = processBuilder.start();

            // 读取标准输出
            BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
            String line;
            while ((line = reader.readLine()) != null) {
                System.out.println("Output: " + line);
            }

            // 读取标准错误
            BufferedReader errorReader = new BufferedReader(new InputStreamReader(process.getErrorStream()));
            while ((line = errorReader.readLine()) != null) {
                System.err.println("Error: " + line);
            }

            int exitCode = process.waitFor(); // 等待命令执行完成
            System.out.println("Exited with error code : " + exitCode);

        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        executeCommand("ls -l /tmp");
    }
}

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.IOException;
import java.util.concurrent.*;
import java.util.function.Consumer;

public class ConcurrentCommandExecutor {

    // 建议根据系统核心数和任务特性调整线程池大小
    private static final int MAX_CONCURRENT_COMMANDS = 100; 
    private final ExecutorService executorService;

    public ConcurrentCommandExecutor() {
        // 创建一个固定大小的线程池，用于执行命令任务
        this.executorService = Executors.newFixedThreadPool(MAX_CONCURRENT_COMMANDS);
    }

    /**
     * 提交一个命令到线程池执行
     * @param command 要执行的Linux命令
     * @param outputConsumer 用于处理标准输出的消费者
     * @param errorConsumer 用于处理标准错误的消费者
     * @return Future对象，可用于获取命令执行结果或等待完成
     */
    public Future<Integer> submitCommand(String command, 
                                         Consumer<String> outputConsumer, 
                                         Consumer<String> errorConsumer) {
        return executorService.submit(() -> {
            ProcessBuilder processBuilder = new ProcessBuilder();
            processBuilder.command("bash", "-c", command);
            processBuilder.redirectErrorStream(false); // 明确区分标准输出和标准错误

            Process process = null;
            int exitCode = -1;
            try {
                process = processBuilder.start();

                // 异步读取标准输出和标准错误，防止阻塞
                CompletableFuture<Void> outputFuture = CompletableFuture.runAsync(() -> {
                    try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
                        String line;
                        while ((line = reader.readLine()) != null) {
                            outputConsumer.accept(line);
                        }
                    } catch (IOException e) {
                        System.err.println("Error reading stdout for command: " + command + ", " + e.getMessage());
                    }
                });

                CompletableFuture<Void> errorFuture = CompletableFuture.runAsync(() -> {
                    try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
                        String line;
                        while ((line = reader.readLine()) != null) {
                            errorConsumer.accept(line);
                        }
                    } catch (IOException e) {
                        System.err.println("Error reading stderr for command: " + command + ", " + e.getMessage());
                    }
                });

                // 等待子进程完成
                exitCode = process.waitFor();

                // 确保所有I/O流都已处理完毕
                outputFuture.join(); 
                errorFuture.join();

            } catch (IOException | InterruptedException e) {
                System.err.println("Failed to execute command: " + command + ", " + e.getMessage());
                if (e instanceof InterruptedException) {
                    Thread.currentThread().interrupt(); // 重新设置中断标志
                }
            } finally {
                if (process != null) {
                    process.destroy(); // 确保进程被终止
                }
            }
            return exitCode;
        });
    }

    /**
     * 关闭线程池
     */
    public void shutdown() {
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow(); // 强制关闭
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ConcurrentCommandExecutor executor = new ConcurrentCommandExecutor();
        int numCommands = 5000; // 模拟执行5000个socat命令
        CountDownLatch latch = new CountDownLatch(numCommands);

        System.out.println("Starting to submit " + numCommands + " commands...");
        long startTime = System.currentTimeMillis();

        for (int i = 0; i < numCommands; i++) {
            final int commandId = i;
            // 假设socat命令不产生大量输出，或者输出不需实时解析
            // 这里的socat命令仅为示例，实际应替换为有意义的命令
            String socatCommand = String.format("socat TCP-LISTEN:%d,fork TCP:127.0.0.1:80 &", 8000 + i); 
            // 注意：真实场景中，socat通常作为守护进程运行，这里只是模拟其启动
            // 如果socat需要长时间运行，需要更复杂的生命周期管理，例如记录其PID以便后续kill

            executor.submitCommand(socatCommand,
                output -> { /* System.out.println("CMD " + commandId + " Output: " + output); */ }, // 避免大量输出导致性能问题
                error -> { System.err.println("CMD " + commandId + " Error: " + error); }
            ).whenComplete((exitCode, throwable) -> {
                if (throwable == null) {
                    // System.out.println("Command " + commandId + " finished with exit code: " + exitCode);
                } else {
                    System.err.println("Command " + commandId + " failed: " + throwable.getMessage());
                }
                latch.countDown();
            });
        }

        latch.await(); // 等待所有命令完成
        long endTime = System.currentTimeMillis();
        System.out.println("All " + numCommands + " commands finished in " + (endTime - startTime) + " ms.");

        executor.shutdown();
        System.out.println("Executor service shut down.");
    }
}