ProcessBuilder: Przekazywanie stdout i stderr uruchomionych procesów bez blokowania głównego wątku

Buduję proces w Javie przy użyciu ProcessBuilder w następujący sposób:

ProcessBuilder pb = new ProcessBuilder()
        .command("somecommand", "arg1", "arg2")
        .redirectErrorStream(true);
Process p = pb.start();

InputStream stdOut = p.getInputStream();

Teraz mój problem jest następujący: chciałbym przechwycić wszystko, co przechodzi przez standardowe wyjście i / lub stderr tego procesu i przekierować to na System.outasynchronicznie. Chcę, aby proces i jego przekierowanie wyjścia działały w tle. Jak dotąd, jedynym sposobem, w jaki udało mi się to zrobić, jest ręczne utworzenie nowego wątku, który będzie w sposób ciągły odczytywał, stdOuta następnie wywoływał odpowiednią write()metodę System.out.

new Thread(new Runnable(){
    public void run(){
        byte[] buffer = new byte[8192];
        int len = -1;
        while((len = stdOut.read(buffer)) > 0){
            System.out.write(buffer, 0, len);
        }
    }
}).start();

Chociaż takie podejście działa, wydaje się trochę brudne. A do tego daje mi jeszcze jeden wątek do prawidłowego zarządzania i zakończenia. Czy jest lepszy sposób, aby to zrobić?

Jedynym sposobem w Javie 6 lub starszym jest tzw. StreamGobbler(Który zaczynasz tworzyć):

StreamGobbler errorGobbler = new StreamGobbler(p.getErrorStream(), "ERROR");

// any output?
StreamGobbler outputGobbler = new StreamGobbler(p.getInputStream(), "OUTPUT");

// start gobblers
outputGobbler.start();
errorGobbler.start();

...

private class StreamGobbler extends Thread {
    InputStream is;
    String type;

    private StreamGobbler(InputStream is, String type) {
        this.is = is;
        this.type = type;
    }

    @Override
    public void run() {
        try {
            InputStreamReader isr = new InputStreamReader(is);
            BufferedReader br = new BufferedReader(isr);
            String line = null;
            while ((line = br.readLine()) != null)
                System.out.println(type + "> " + line);
        }
        catch (IOException ioe) {
            ioe.printStackTrace();
        }
    }
}

W przypadku Java 7 zobacz odpowiedź Evgeniy Dorofeev.

Użyj ProcessBuilder.inheritIO, ustawia źródło i miejsce docelowe dla standardowego wejścia / wyjścia podprocesu na takie same, jak te z bieżącego procesu Java.

Process p = new ProcessBuilder().inheritIO().command("command1").start();

Jeśli Java 7 nie jest opcją

public static void main(String[] args) throws Exception {
    Process p = Runtime.getRuntime().exec("cmd /c dir");
    inheritIO(p.getInputStream(), System.out);
    inheritIO(p.getErrorStream(), System.err);

}

private static void inheritIO(final InputStream src, final PrintStream dest) {
    new Thread(new Runnable() {
        public void run() {
            Scanner sc = new Scanner(src);
            while (sc.hasNextLine()) {
                dest.println(sc.nextLine());
            }
        }
    }).start();
}

Wątki umrą automatycznie po zakończeniu podprocesu, ponieważ srcnastąpi EOF.

Elastyczne rozwiązanie z lambdą Java 8, które pozwala zapewnić Consumerprzetwarzanie danych wyjściowych (np. Rejestrowanie) wiersz po wierszu. run()jest jednym wierszem bez rzuconych zaznaczonych wyjątków. Alternatywnie do implementacji Runnable, może się rozszerzyć, Threadjak sugerują inne odpowiedzi.

class StreamGobbler implements Runnable {
    private InputStream inputStream;
    private Consumer<String> consumeInputLine;

    public StreamGobbler(InputStream inputStream, Consumer<String> consumeInputLine) {
        this.inputStream = inputStream;
        this.consumeInputLine = consumeInputLine;
    }

    public void run() {
        new BufferedReader(new InputStreamReader(inputStream)).lines().forEach(consumeInputLine);
    }
}

Możesz wtedy użyć go na przykład w ten sposób:

public void runProcessWithGobblers() throws IOException, InterruptedException {
    Process p = new ProcessBuilder("...").start();
    Logger logger = LoggerFactory.getLogger(getClass());

    StreamGobbler outputGobbler = new StreamGobbler(p.getInputStream(), System.out::println);
    StreamGobbler errorGobbler = new StreamGobbler(p.getErrorStream(), logger::error);

    new Thread(outputGobbler).start();
    new Thread(errorGobbler).start();
    p.waitFor();
}

Tutaj strumień wyjściowy jest przekierowywany, System.outa strumień błędów jest rejestrowany na poziomie błędu przez logger.

To jest tak proste, jak następujące:

    File logFile = new File(...);
    ProcessBuilder pb = new ProcessBuilder()
        .command("somecommand", "arg1", "arg2")
    processBuilder.redirectErrorStream(true);
    processBuilder.redirectOutput(logFile);

przez .redirectErrorStream (true) mówisz procesowi, aby scalił błąd i strumień wyjściowy, a następnie przez .redirectOutput (plik) przekierowujesz scalone dane wyjściowe do pliku.

Aktualizacja:

Udało mi się to zrobić w następujący sposób:

public static void main(String[] args) {
    // Async part
    Runnable r = () -> {
        ProcessBuilder pb = new ProcessBuilder().command("...");
        // Merge System.err and System.out
        pb.redirectErrorStream(true);
        // Inherit System.out as redirect output stream
        pb.redirectOutput(ProcessBuilder.Redirect.INHERIT);
        try {
            pb.start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    };
    new Thread(r, "asyncOut").start();
    // here goes your main part
}

Teraz możesz zobaczyć zarówno dane wyjściowe z wątków głównych, jak i asynchronicznych w System.out

Proste rozwiązanie java8 z przechwytywaniem obu wyjść i reaktywnym przetwarzaniem za pomocą CompletableFuture:

static CompletableFuture<String> readOutStream(InputStream is) {
    return CompletableFuture.supplyAsync(() -> {
        try (
                InputStreamReader isr = new InputStreamReader(is);
                BufferedReader br = new BufferedReader(isr);
        ){
            StringBuilder res = new StringBuilder();
            String inputLine;
            while ((inputLine = br.readLine()) != null) {
                res.append(inputLine).append(System.lineSeparator());
            }
            return res.toString();
        } catch (Throwable e) {
            throw new RuntimeException("problem with executing program", e);
        }
    });
}

I użycie:

Process p = Runtime.getRuntime().exec(cmd);
CompletableFuture<String> soutFut = readOutStream(p.getInputStream());
CompletableFuture<String> serrFut = readOutStream(p.getErrorStream());
CompletableFuture<String> resultFut = soutFut.thenCombine(serrFut, (stdout, stderr) -> {
         // print to current stderr the stderr of process and return the stdout
        System.err.println(stderr);
        return stdout;
        });
// get stdout once ready, blocking
String result = resultFut.get();

Istnieje biblioteka, która zapewnia lepszy ProcessBuilder, zt-exec. Ta biblioteka może zrobić dokładnie to, o co prosisz, a nawet więcej.

Oto jak wyglądałby Twój kod z zt-exec zamiast ProcessBuilder:

dodaj zależność:

<dependency>
  <groupId>org.zeroturnaround</groupId>
  <artifactId>zt-exec</artifactId>
  <version>1.11</version>
</dependency>

Kod :

new ProcessExecutor()
  .command("somecommand", "arg1", "arg2")
  .redirectOutput(System.out)
  .redirectError(System.err)
  .execute();

Dokumentacja biblioteki jest tutaj: https://github.com/zeroturnaround/zt-exec/

Ja też mogę używać tylko Javy 6. Użyłem implementacji skanera wątków @ EvgeniyDorofeev. W moim kodzie, po zakończeniu procesu, muszę natychmiast wykonać dwa inne procesy, z których każdy porównuje przekierowane dane wyjściowe (test jednostkowy oparty na diff, aby upewnić się, że stdout i stderr są takie same jak błogosławione).

Wątki skanera nie kończą się wystarczająco szybko, nawet jeśli proces waitFor () zostanie zakończony. Aby kod działał poprawnie, muszę upewnić się, że wątki są połączone po zakończeniu procesu.

public static int runRedirect (String[] args, String stdout_redirect_to, String stderr_redirect_to) throws IOException, InterruptedException {
    ProcessBuilder b = new ProcessBuilder().command(args);
    Process p = b.start();
    Thread ot = null;
    PrintStream out = null;
    if (stdout_redirect_to != null) {
        out = new PrintStream(new BufferedOutputStream(new FileOutputStream(stdout_redirect_to)));
        ot = inheritIO(p.getInputStream(), out);
        ot.start();
    }
    Thread et = null;
    PrintStream err = null;
    if (stderr_redirect_to != null) {
        err = new PrintStream(new BufferedOutputStream(new FileOutputStream(stderr_redirect_to)));
        et = inheritIO(p.getErrorStream(), err);
        et.start();
    }
    p.waitFor();    // ensure the process finishes before proceeding
    if (ot != null)
        ot.join();  // ensure the thread finishes before proceeding
    if (et != null)
        et.join();  // ensure the thread finishes before proceeding
    int rc = p.exitValue();
    return rc;
}

private static Thread inheritIO (final InputStream src, final PrintStream dest) {
    return new Thread(new Runnable() {
        public void run() {
            Scanner sc = new Scanner(src);
            while (sc.hasNextLine())
                dest.println(sc.nextLine());
            dest.flush();
        }
    });
}

Jako dodatek do odpowiedzi msangel chciałbym dodać następujący blok kodu:

private static CompletableFuture<Boolean> redirectToLogger(final InputStream inputStream, final Consumer<String> logLineConsumer) {
        return CompletableFuture.supplyAsync(() -> {
            try (
                InputStreamReader inputStreamReader = new InputStreamReader(inputStream);
                BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
            ) {
                String line = null;
                while((line = bufferedReader.readLine()) != null) {
                    logLineConsumer.accept(line);
                }
                return true;
            } catch (IOException e) {
                return false;
            }
        });
    }

Pozwala na przekierowanie strumienia wejściowego (stdout, stderr) procesu na innego konsumenta. Może to być System.out :: println lub cokolwiek innego pochłaniającego łańcuchy.

Stosowanie:

...
Process process = processBuilder.start()
CompletableFuture<Boolean> stdOutRes = redirectToLogger(process.getInputStream(), System.out::println);
CompletableFuture<Boolean> stdErrRes = redirectToLogger(process.getErrorStream(), System.out::println);
System.out.println(stdOutRes.get());
System.out.println(stdErrRes.get());
System.out.println(process.waitFor());

Thread thread = new Thread(() -> {
      new BufferedReader(
          new InputStreamReader(inputStream, 
                                StandardCharsets.UTF_8))
              .lines().forEach(...);
    });
    thread.start();

Twój kod niestandardowy idzie zamiast ...

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

public class Main {

    public static void main(String[] args) throws Exception {
        ProcessBuilder pb = new ProcessBuilder("script.bat");
        pb.redirectErrorStream(true);
        Process p = pb.start();
        BufferedReader logReader = new BufferedReader(new InputStreamReader(p.getInputStream()));
        String logLine = null;
        while ( (logLine = logReader.readLine()) != null) {
           System.out.println("Script output: " + logLine);
        }
    }
}

Używając tej linii: pb.redirectErrorStream(true);możemy połączyć InputStream i ErrorStream

Domyślnie utworzony podproces nie ma własnego terminala ani konsoli. Wszystkie jego standardowe operacje we / wy (tj. Stdin, stdout, stderr) zostaną przekierowane do procesu nadrzędnego, gdzie można uzyskać do nich dostęp za pośrednictwem strumieni uzyskanych za pomocą metod getOutputStream (), getInputStream () i getErrorStream (). Proces nadrzędny wykorzystuje te strumienie do dostarczania danych wejściowych do podprocesu i pobierania z niego danych wyjściowych. Ponieważ niektóre platformy natywne zapewniają ograniczony rozmiar buforu tylko dla standardowych strumieni wejściowych i wyjściowych, niepowodzenie szybkiego zapisania strumienia wejściowego lub odczytu strumienia wyjściowego podprocesu może spowodować zablokowanie lub nawet zakleszczenie podprocesu.

https://www.securecoding.cert.org/confluence/display/java/FIO07-J.+Do+not+let+external+processes+block+on+IO+buffers