ProcessBuilder: Meneruskan stdout dan stderr proses yang dimulai tanpa memblokir thread utama

Saya sedang membangun proses di Java menggunakan ProcessBuilder sebagai berikut:

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

InputStream stdOut = p.getInputStream();

Sekarang masalah saya adalah sebagai berikut: Saya ingin menangkap apa pun yang terjadi melalui stdout dan / atau stderr dari proses itu dan mengarahkannya ke System.outasynchronous. Saya ingin proses dan pengalihan keluarannya berjalan di latar belakang. Sejauh ini, satu-satunya cara yang saya temukan untuk melakukan ini adalah dengan menelurkan utas baru secara manual yang akan terus membaca stdOutdan kemudian memanggil write()metode yang sesuai dari 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();

Sementara pendekatan semacam itu berhasil, rasanya agak kotor. Dan yang terpenting, ini memberi saya satu utas lagi untuk dikelola dan diakhiri dengan benar. Apakah ada cara yang lebih baik untuk melakukan ini?

Untuk satu-satunya cara di Java 6 atau sebelumnya adalah dengan apa yang disebut StreamGobbler(yang Anda mulai buat):

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

Untuk Java 7, lihat jawaban Evgeniy Dorofeev.

Gunakan ProcessBuilder.inheritIO, ini menetapkan sumber dan tujuan untuk subproses standar I / O menjadi sama dengan proses Java saat ini.

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

Jika Java 7 bukan pilihan

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

Benang akan mati secara otomatis saat subproses selesai, karena srcakan EOF.

Solusi fleksibel dengan Java 8 lambda yang memungkinkan Anda menyediakan Consumeryang akan memproses keluaran (mis. Mencatatnya) baris demi baris. run()adalah satu baris tanpa ada pengecualian yang dicentang. Sebagai alternatif untuk menerapkan Runnable, itu dapat diperluas Threadsebagai gantinya seperti yang disarankan oleh jawaban lain.

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

Anda kemudian dapat menggunakannya sebagai contoh seperti ini:

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

Di sini aliran keluaran diarahkan ke System.outdan aliran kesalahan dicatat pada tingkat kesalahan oleh logger.

Sesederhana berikut:

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

oleh .redirectErrorStream (true) Anda memberi tahu proses untuk menggabungkan kesalahan dan aliran keluaran dan kemudian dengan .redirectOutput (file) Anda mengarahkan keluaran gabungan ke file.

Memperbarui:

Saya berhasil melakukan ini sebagai berikut:

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
}

Sekarang Anda dapat melihat kedua keluaran dari utas main dan asyncOut di System.out

Solusi java8 sederhana dengan menangkap keluaran dan pemrosesan reaktif menggunakan 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);
        }
    });
}

Dan penggunaannya:

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

Ada pustaka yang menyediakan ProcessBuilder yang lebih baik, zt-exec. Perpustakaan ini dapat melakukan apa yang Anda minta dan banyak lagi.

Inilah tampilan kode Anda dengan zt-exec dan bukan ProcessBuilder:

tambahkan ketergantungan:

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

Kode :

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

Dokumentasi perpustakaan ada di sini: https://github.com/zeroturnaround/zt-exec/

Saya juga hanya dapat menggunakan Java 6. Saya menggunakan implementasi pemindai thread @ EvgeniyDorofeev. Dalam kode saya, setelah proses selesai, saya harus segera menjalankan dua proses lain yang masing-masing membandingkan keluaran yang dialihkan (pengujian unit berbasis diff untuk memastikan stdout dan stderr sama dengan yang diberkati).

Untaian pemindai tidak segera selesai, bahkan jika saya menunggu () prosesnya selesai. Untuk membuat kode berfungsi dengan benar, saya harus memastikan utas bergabung setelah proses selesai.

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

Sebagai tambahan untuk jawaban msangel saya ingin menambahkan blok kode berikut:

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

Ini memungkinkan untuk mengarahkan aliran input (stdout, stderr) dari proses ke beberapa konsumen lain. Ini mungkin System.out :: println atau apa pun yang menggunakan string.

Pemakaian:

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

Kode khusus Anda bukan ...

Secara default, subproses yang dibuat tidak memiliki terminal atau konsolnya sendiri. Semua operasi I / O standarnya (yaitu stdin, stdout, stderr) akan diarahkan ke proses induk, di mana mereka dapat diakses melalui aliran yang diperoleh menggunakan metode getOutputStream (), getInputStream (), dan getErrorStream (). Proses induk menggunakan aliran ini untuk memberi masukan dan mendapatkan keluaran dari subproses. Karena beberapa platform asli hanya menyediakan ukuran buffer terbatas untuk aliran input dan output standar, kegagalan untuk segera menulis aliran input atau membaca aliran output dari subproses dapat menyebabkan subproses diblokir, atau bahkan kebuntuan.

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