ProcessBuilder: reenvío stdout y stderr de procesos iniciados sin bloquear el hilo principal

Estoy construyendo un proceso en Java usando ProcessBuilder de la siguiente manera:

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

InputStream stdOut = p.getInputStream();

Ahora mi problema es el siguiente: me gustaría capturar lo que esté pasando por stdout y / o stderr de ese proceso y redirigirlo de forma System.outasincrónica. Quiero que el proceso y su redirección de salida se ejecuten en segundo plano. Hasta ahora, la única forma que he encontrado para hacer esto es generar manualmente un nuevo hilo que leerá continuamente stdOuty luego llamará al write()método apropiado de 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();

Si bien ese enfoque funciona, se siente un poco sucio. Y además de eso, me da un hilo más para administrar y terminar correctamente. ¿Hay alguna forma mejor de hacer esto?

La única forma en Java 6 o anterior es con un llamado StreamGobbler(que ya está empezando a crear):

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();
        }
    }
}

Para Java 7, consulte la respuesta de Evgeniy Dorofeev.

Utilícelo ProcessBuilder.inheritIO, establece el origen y el destino de la E / S estándar del subproceso para que sean los mismos que los del proceso Java actual.

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

Si Java 7 no es una opción

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();
}

Los subprocesos morirán automáticamente cuando finalice el subproceso, porque lo srchará EOF.

Una solución flexible con Java 8 lambda que le permite proporcionar una Consumerque procesará la salida (por ejemplo, registrarla) línea por línea. run()es una línea de una sola línea sin excepciones marcadas lanzadas. Alternativamente a la implementación Runnable, puede extenderse Threadcomo sugieren otras respuestas.

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);
    }
}

Luego puede usarlo, por ejemplo, así:

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();
}

Aquí, el flujo de salida se redirige a System.outy el flujo de error se registra en el nivel de error mediante logger.

Es tan simple como seguir:

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

por .redirectErrorStream (verdadero) le dice al proceso que combine el error y el flujo de salida y luego por .redirectOutput (archivo) redirige la salida combinada a un archivo.

Actualizar:

Me las arreglé para hacer esto de la siguiente manera:

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
}

Ahora puede ver las salidas de los subprocesos principal y asyncOut en System.out

Solución java8 simple con captura de ambas salidas y procesamiento reactivo usando 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);
        }
    });
}

Y el uso:

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();

Hay una biblioteca que proporciona un ProcessBuilder mejor, zt-exec. Esta biblioteca puede hacer exactamente lo que está pidiendo y más.

Así es como se vería su código con zt-exec en lugar de ProcessBuilder:

agregue la dependencia:

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

El código :

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

La documentación de la biblioteca está aquí: https://github.com/zeroturnaround/zt-exec/

Yo también puedo usar solo Java 6. Usé la implementación del escáner de subprocesos de @ EvgeniyDorofeev. En mi código, después de que finaliza un proceso, tengo que ejecutar inmediatamente otros dos procesos que comparan la salida redirigida (una prueba unitaria basada en diferencias para garantizar que stdout y stderr sean los mismos que los bendecidos).

Los subprocesos del escáner no terminan lo suficientemente pronto, incluso si espero () que se complete el proceso. Para que el código funcione correctamente, tengo que asegurarme de que los hilos se unan después de que finalice el proceso.

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();
        }
    });
}

Como adición a la respuesta de msangel, me gustaría agregar el siguiente bloque de código:

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;
            }
        });
    }

Permite redirigir el flujo de entrada (stdout, stderr) del proceso a algún otro consumidor. Este podría ser System.out :: println o cualquier otra cosa que consuma cadenas.

Uso:

...
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();

Su código personalizado va en lugar del ...

De forma predeterminada, el subproceso creado no tiene su propia terminal o consola. Todas sus operaciones de E / S estándar (es decir, stdin, stdout, stderr) serán redirigidas al proceso padre, donde se puede acceder a ellas a través de las secuencias obtenidas mediante los métodos getOutputStream (), getInputStream () y getErrorStream (). El proceso padre utiliza estos flujos para alimentar la entrada y obtener la salida del subproceso. Debido a que algunas plataformas nativas solo proporcionan un tamaño de búfer limitado para los flujos de entrada y salida estándar, si no se escribe rápidamente el flujo de entrada o se lee el flujo de salida del subproceso, es posible que el subproceso se bloquee o incluso se bloquee.

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