Como posso lidar com múltiplos fluxos em Java?

Question

Eu estou tentando executar um processo e fazer coisas com seus fluxos de entrada, saída e de erro. A maneira óbvia de fazer isso é usar algo como select(), mas a única coisa que posso encontrar em Java que faz isso é Selector.select(), que leva um Channel. Não parece ser possível obter um Channel de um InputStream ou OutputStream (FileStream tem um método getChannel() mas isso não ajudar aqui)

Assim, em vez eu escrevi algum código para consultar todos os fluxos:

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, exceto que ele nunca termina. Quando um dos córregos chegue ao fim de arquivo, available() retorna zero para read() não é chamado e nós nunca chegar a -1 retorno que poderia indicar EOF.

Uma solução seria uma maneira non-blocking para detectar EOF. Eu não posso ver um nos docs em qualquer lugar. Como alternativa há uma maneira melhor de fazer o que eu quero fazer?

Eu vejo esta pergunta aqui: texto do link e apesar de não fazer exatamente o que eu quero, eu provavelmente pode usar essa idéia, de desova segmentos separados para cada fluxo, para o problema particular que tenho agora. Mas, certamente, isso não é a única maneira de fazê-lo? Certamente deve haver uma maneira de ler a partir de múltiplos fluxos sem o uso de um fio para cada um?

Answer 1

Como você disse, a solução descrito neste resposta é a forma tradicional de leitura de ambos stdout e stderr de um processo. Um fio-per-stream é o caminho a percorrer, embora seja um pouco irritante.

Answer 2

Você de fato terá que ir a rota de gerando uma thread para cada fluxo você deseja monitorar. Se o seu caso de uso permite a combinação de ambos stdout e stderr do processo em questão que você precisa apenas um segmento, caso contrário, dois são necessários.

Levei algum tempo para obtê-lo direito em um de nossos projetos em que eu tenho que lançar um processo externo, tome a sua saída e fazer algo com ele, enquanto ao mesmo tempo à procura de erros e encerramento do processo e também ser capaz de denunciá-lo quando o usuário do aplicativo java cancela a operação.

Eu criei uma classe bastante simples para encapsular a parte assistindo cujo método run () é algo como isto:

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

No lado chamando-o parecido com este:

    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;

Infelizmente eu não tenho que para um exemplo executável auto-suficiente, mas talvez isso ajude. Se eu tivesse que fazê-lo novamente eu teria um outro olhar para usando o método Thread.interrupt () em vez de uma parada bandeira self-made (mente, declarando-volátil!), Mas deixo isso para outra altura. :)