如何通过ProcessBuilder将字符串列表作为stdin传入系统命令？

需要实现等效于echo -e "foo\nbar\nspam" | grep 'foo'的功能——将字符串列表逐行作为外部命令的stdin输入，避免使用临时文件，但当前用ProcessBuilder遇到大数据量阻塞、结果数量不稳定的问题。

初始代码草稿 #

尝试寻找直接传入列表的API，但ProcessBuilder无此功能：

List<String> input = new ArrayList<String>();
input.add("foo");
input.add("bar");
input.add("spam");

List<String> cmd = new ArrayList<String>();
cmd.add("grep");
cmd.add("foo");

ProcessBuilder pb = new ProcessBuilder().command(cmd);

// *** 此处需要将input传入ProcessBuilder，例如：***
// pb.inputFromList(input);

Process p= pb.start();
BufferedReader reader= new BufferedReader(new InputStreamReader(p.getInputStream()));
// 处理仅包含"foo"的reader内容。

更新1：小数据正常，大数据量阻塞 #

参考方案直接向进程输出流写入数据，但输入列表达到数万条时出现阻塞：

List<String> inputList = new ArrayList<String>();
for(int i = 0; i < 100; i++) {
    // 小数据正常，i>50000时会阻塞
    inputList.add("foo");
    inputList.add("bar");
}

List<String> cmd = new ArrayList<String>();
cmd.add("grep");
cmd.add("foo");

ProcessBuilder pb = new ProcessBuilder().command(cmd);
Process p = pb.start();

OutputStream stdin = p.getOutputStream();

int i = 0;
for(String line : inputList){
    stdin.write(line.getBytes(StandardCharsets.UTF_8));
    stdin.write('\n');
    i++;
    System.err.println(i);
}
stdin.close();

BufferedReader reader= new BufferedReader(new InputStreamReader(p.getInputStream()));

String line = "";
while ((line = reader.readLine())!= null) {
    System.err.println(line);
}
reader.close();
p.waitFor();

问题：添加pb.redirectInput(ProcessBuilder.Redirect.INHERIT)会抛出java.io.IOException: Stream closed，不添加则大数据量时阻塞。

更新2：结果数量不稳定 #

尝试用线程读取输出，但多次运行返回的结果数量不一致：

public static ArrayList<String> execSystemCommand(List<String> inputList, List<String> cmd) throws IOException, InterruptedException {

    ProcessBuilder pb = new ProcessBuilder().command(cmd);
    pb.redirectErrorStream(true); // 是否需要此行？
    Process p = pb.start();

    ArrayList<String> results= new ArrayList<String>();
    
    Thread readerThread = new Thread(() -> {
        try {
            try (BufferedReader reader = new BufferedReader(new 
                    InputStreamReader(p.getInputStream()))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    results.add(line);
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("Unhandled", e);
        }
    });
    readerThread.start();
    
    OutputStream stdin = p.getOutputStream();

    for(String line : inputList){
        stdin.write(line.getBytes(StandardCharsets.UTF_8));
        stdin.write('\n');
    }
    stdin.close();
    p.waitFor();
    
    return results;
}

测试用例：

@Test
public void test() throws IOException, InterruptedException {
    
    // 输入列表为[foo0, foo1, ..., foo99999]
    List<String> inputList = new ArrayList<String>();
    for(int i = 0; i < 100000; i++) {
        inputList.add("foo" + i);
    }
    
    List<String> cmd = new ArrayList<String>();
    cmd.add("grep");
    cmd.add("foo");

    ArrayList<String> results = Utils.execSystemCommand(inputList, cmd);
    
    System.out.println("Size:" + results.size());
    System.out.println("First element:" + results.get(0));
    System.out.println("Last element:" + results.get(results.size() - 1));
}

执行结果示例：

Size:93166
First element:foo0
Last element:foo93165

Size:100000
First element:foo0
Last element:foo99999

Size:93622
First element:foo0
Last element:foo96351

问题原因分析 #

1. 数据流阻塞：进程的输入、输出流都有缓冲区，当缓冲区满时，读写操作会阻塞。若主线程仅负责写入输入并等待进程结束，进程输出缓冲区满时无法继续处理输入，会导致死锁。
2. 结果收集不完整：更新2中启动读取线程后，p.waitFor()直接返回结果，此时读取线程可能还未完成全部输出的读取，导致结果数量不稳定。
3. 错误流未处理：若不合并错误流，错误流缓冲区满也会导致进程阻塞。

修正方案 #

需要独立线程处理输入写入和输出读取，确保所有数据流被及时消费，且等待所有线程完成后再返回结果。使用缓冲流提升效率，避免阻塞。

修正后的代码 #

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

public class ProcessUtils {

    public static List<String> execSystemCommand(List<String> inputList, List<String> cmd) throws IOException, InterruptedException {
        ProcessBuilder pb = new ProcessBuilder(cmd);
        pb.redirectErrorStream(true); // 合并错误流到输出流，避免错误缓冲区满导致阻塞
        Process process = pb.start();

        List<String> results = new ArrayList<>();

        // 线程1：写入输入
        Thread writeThread = new Thread(() -> {
            try (BufferedWriter writer = new BufferedWriter(
                    new OutputStreamWriter(process.getOutputStream(), StandardCharsets.UTF_8))) {
                for (String line : inputList) {
                    writer.write(line);
                    writer.newLine();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        writeThread.start();

        // 线程2：读取输出
        Thread readThread = new Thread(() -> {
            try (BufferedReader reader = new BufferedReader(
                    new InputStreamReader(process.getInputStream(), StandardCharsets.UTF_8))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    synchronized (results) {
                        results.add(line);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        readThread.start();

        // 等待输入写入完成
        writeThread.join();
        // 等待进程执行完成
        int exitCode = process.waitFor();
        // 等待输出读取完成
        readThread.join();

        if (exitCode != 0) {
            System.err.println("Command exited with code: " + exitCode);
        }

        return results;
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        List<String> inputList = new ArrayList<>();
        for (int i = 0; i < 100000; i++) {
            inputList.add("foo" + i);
        }

        List<String> cmd = new ArrayList<>();
        cmd.add("grep");
        cmd.add("foo");

        List<String> results = execSystemCommand(inputList, cmd);

        System.out.println("Size:" + results.size());
        System.out.println("First element:" + results.get(0));
        System.out.println("Last element:" + results.get(results.size() - 1));
    }
}

关键优化点 #

1. 缓冲流提升效率：BufferedWriter和BufferedReader减少IO操作次数，提升大数据量处理速度。
2. 独立线程处理读写：输入写入和输出读取分开执行，避免互相阻塞导致死锁。
3. 等待线程完成：writeThread.join()确保输入全部写入，readThread.join()确保所有输出被收集，避免结果不完整。
4. 合并错误流：防止错误流缓冲区满导致进程阻塞，同时错误信息会被包含在结果中（若需单独处理可取消此设置）。
5. 线程安全收集结果：对结果列表的添加操作加锁，避免多线程并发修改导致的异常或数据丢失。

内容的提问来源于stack exchange，提问作者dariober