¿Cómo manejo múltiples transmisiones en Java?

Question

Estoy intentando ejecutar un proceso y hacer cosas con sus flujos de entrada, salida y error.La forma obvia de hacer esto es usar algo como select(), pero lo único que puedo encontrar en Java que hace eso es Selector.select(), que toma un Channel.No parece posible conseguir una Channel desde un InputStream o OutputStream (FileStream tiene un getChannel() método pero eso no ayuda aquí)

Entonces, en lugar de eso, escribí un código para sondear todas las transmisiones:

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

que funciona, excepto que nunca termina.Cuando una de las secuencias llega al final del archivo, available() devuelve cero entonces read() no se llama y nunca obtenemos el retorno -1 que indicaría EOF.

Una solución sería una forma sin bloqueo de detectar EOF.No puedo ver uno en los documentos en ninguna parte.Alternativamente, ¿hay una mejor manera de hacer lo que quiero hacer?

Veo esta pregunta aquí:Texto del enlacey aunque no hace exactamente lo que quiero, probablemente pueda usar esa idea, de generar hilos separados para cada flujo, para el problema particular que tengo ahora.¿Pero seguramente esa no es la única manera de hacerlo?Seguramente debe haber una manera de leer desde múltiples transmisiones sin usar un hilo para cada una.

Answer 1

Como dijiste, la solución. descrito en esta respuesta es la forma tradicional de leer tanto stdout como stderr de un Proceso.Un hilo por flujo es el camino a seguir, aunque sea un poco molesto.

Answer 2

De hecho, tendrás que seguir la ruta de generar un hilo para cada transmisión que desees monitorear.Si su caso de uso permite combinar stdout y stderr del proceso en cuestión, solo necesita un subproceso; de lo contrario, se necesitan dos.

Me tomó bastante tiempo hacerlo bien en uno de nuestros proyectos donde tengo que iniciar un proceso externo, tomar su resultado y hacer algo con él mientras al mismo tiempo busco errores y terminación del proceso y también puedo terminarlo. cuando el usuario de la aplicación java cancela la operación.

Creé una clase bastante simple para encapsular la parte de observación cuyo método run() se parece a esto:

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

En el lado de la llamada se ve así:

    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;

Desafortunadamente, no es necesario para un ejemplo ejecutable autónomo, pero tal vez esto ayude.Si tuviera que hacerlo de nuevo, volvería a intentar usar el método Thread.interrupt() en lugar de una bandera de parada hecha por mí mismo (¡tenga cuidado de declararlo volátil!), pero lo dejo para otro momento.:)