Come gestisco più stream in Java?

Question

Sto cercando di eseguire un processo e fare cose con i suoi flussi di input, output ed errori. Il modo ovvio per farlo è usare qualcosa come select () , ma l'unica cosa che posso trovare in Java che lo fa è Selector.select () , che richiede un canale . Non sembra possibile ottenere un canale da un InputStream o OutputStream ( FileStream ha un < codice> getChannel () ma questo non aiuta qui)

Quindi, invece, ho scritto del codice per eseguire il polling di tutti i flussi:

while( !out_eof || !err_eof )
{
    while( out_str.available() )
    {
        if( (bytes = out_str.read(buf)) != -1 )
        {
            // Do something with output stream
        }
        else
            out_eof = true;
    }
    while( err_str.available() )
    {
        if( (bytes = err_str.read(buf)) != -1 )
        {
            // Do something with error stream
        }
        else
            err_eof = true;
    }
    sleep(100);
}

che funziona, tranne per il fatto che non termina mai. Quando uno dei flussi raggiunge la fine del file, available () restituisce zero, quindi read () non viene chiamato e non otteniamo mai il ritorno -1 che indicherebbe EOF.

Una soluzione sarebbe un modo non bloccante per rilevare EOF. Non riesco a vederne uno nei documenti da nessuna parte. In alternativa, esiste un modo migliore di fare ciò che voglio fare?

Vedo questa domanda qui: testo del link e anche se non fa esattamente quello che voglio, probabilmente posso usare quell'idea, di generare thread separati per ogni flusso, per il particolare problema che ho ora. Ma sicuramente non è l'unico modo per farlo? Sicuramente ci deve essere un modo per leggere da più stream senza usare un thread per ciascuno?

Answer 1

Come hai detto, la soluzione delineata in questo Risposta è il modo tradizionale di leggere sia stdout che stderr da un processo. Un thread per stream è la strada da percorrere, anche se è leggermente fastidioso.

Answer 2

Dovrai davvero seguire la strada per generare un thread per ogni flusso che desideri monitorare. Se il tuo caso d'uso consente di combinare sia stdout che stderr del processo in questione, hai bisogno di un solo thread, altrimenti ne occorrono due.

Mi ci è voluto un po 'di tempo per farlo nel modo giusto in uno dei nostri progetti in cui dovevo avviare un processo esterno, prendere il suo output e fare qualcosa con esso, cercando allo stesso tempo errori e terminazione del processo e potendo anche per terminarlo quando l'utente dell'app java annulla l'operazione.

Ho creato una classe piuttosto semplice per incapsulare la parte di controllo il cui metodo run () è simile al seguente:

public void run() {
    BufferedReader tStreamReader = null;
    try {
        while (externalCommand == null && !shouldHalt) {
            logger.warning("ExtProcMonitor("
                           + (watchStdErr ? "err" : "out")
                           + ") Sleeping until external command is found");
            Thread.sleep(500);
        }
        if (externalCommand == null) {
            return;
        }
        tStreamReader =
                new BufferedReader(new InputStreamReader(watchStdErr ? externalCommand.getErrorStream()
                        : externalCommand.getInputStream()));
        String tLine;
        while ((tLine = tStreamReader.readLine()) != null) {
            logger.severe(tLine);
            if (filter != null) {
                if (filter.matches(tLine)) {
                    informFilterListeners(tLine);
                    return;
                }
            }
        }
    } catch (IOException e) {
        logger.logExceptionMessage(e, "IOException stderr");
    } catch (InterruptedException e) {
        logger.logExceptionMessage(e, "InterruptedException waiting for external process");
    } finally {
        if (tStreamReader != null) {
            try {
                tStreamReader.close();
            } catch (IOException e) {
                // ignore
            }
        }
    }
}

Sul lato chiamante sembra così:

    Thread tExtMonitorThread = new Thread(new Runnable() {

        public void run() {
            try {
                while (externalCommand == null) {
                    getLogger().warning("Monitor: Sleeping until external command is found");
                    Thread.sleep(500);
                    if (isStopRequested()) {
                        getLogger()
                                .warning("Terminating external process on user request");
                        if (externalCommand != null) {
                            externalCommand.destroy();
                        }
                        return;
                    }
                }
                int tReturnCode = externalCommand.waitFor();
                getLogger().warning("External command exited with code " + tReturnCode);
            } catch (InterruptedException e) {
                getLogger().logExceptionMessage(e, "Interrupted while waiting for external command to exit");
            }
        }
    }, "ExtCommandWaiter");

    ExternalProcessOutputHandlerThread tExtErrThread =
            new ExternalProcessOutputHandlerThread("ExtCommandStdErr", getLogger(), true);
    ExternalProcessOutputHandlerThread tExtOutThread =
            new ExternalProcessOutputHandlerThread("ExtCommandStdOut", getLogger(), true);
    tExtMonitorThread.start();
    tExtOutThread.start();
    tExtErrThread.start();
    tExtErrThread.setFilter(new FilterFunctor() {

        public boolean matches(Object o) {
            String tLine = (String)o;
            return tLine.indexOf("Error") > -1;
        }
    });

    FilterListener tListener = new FilterListener() {
        private boolean abortFlag = false;

        public boolean shouldAbort() {
            return abortFlag;
        }

        public void matched(String aLine) {
            abortFlag = abortFlag || (aLine.indexOf("Error") > -1);
        }

    };

    tExtErrThread.addFilterListener(tListener);
    externalCommand = new ProcessBuilder(aCommand).start();
    tExtErrThread.setProcess(externalCommand);
    try {
        tExtMonitorThread.join();
        tExtErrThread.join();
        tExtOutThread.join();
    } catch (InterruptedException e) {
        // when this happens try to bring the external process down 
        getLogger().severe("Aborted because auf InterruptedException.");
        getLogger().severe("Killing external command...");
        externalCommand.destroy();
        getLogger().severe("External command killed.");
        externalCommand = null;
        return -42;
    }
    int tRetVal = tListener.shouldAbort() ? -44 : externalCommand.exitValue();

    externalCommand = null;
    try {
        getLogger().warning("command exit code: " + tRetVal);
    } catch (IllegalThreadStateException ex) {
        getLogger().warning("command exit code: unknown");
    }
    return tRetVal;

Sfortunatamente non devo fare un esempio eseguibile autonomo, ma forse questo aiuta. Se dovessi farlo di nuovo, darei un'altra occhiata all'utilizzo del metodo Thread.interrupt () invece di un flag di stop fatto da te (ricordati di dichiararlo volatile!), Ma lo lascio per un'altra volta. :)