Comment gérer plusieurs flux en Java?

Question

J'essaie de lancer un processus et de faire des choses avec ses flux d'entrée, de sortie et d'erreur. La façon évidente d’y parvenir est d’utiliser quelque chose comme select () , mais la seule chose que je peux trouver en Java qui le fait est Selector.select () , qui prend un canal . Il ne semble pas possible d'obtenir un Canal à partir d'un InputStream ou d'un OutputStream ( FileStream a un < code> getChannel () , mais cela n’aide pas ici)

Au lieu de cela, j’ai écrit du code pour interroger tous les flux:

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

qui fonctionne, sauf qu'il ne se termine jamais. Lorsque l'un des flux atteint la fin du fichier, available () renvoie zéro, ainsi read () n'est pas appelé et nous n'obtenons jamais la valeur -1 qui indique EOF.

Une solution serait un moyen non bloquant de détecter EOF. Je ne peux en voir un dans la documentation nulle part. Sinon, y a-t-il une meilleure façon de faire ce que je veux faire?

Je vois cette question ici: texte du lien et bien que cela ne fasse pas exactement ce que je veux, je peux probablement utiliser cette idée de créer des threads distincts pour chaque flux, pour le problème particulier que j'ai maintenant. Mais ce n’est sûrement pas la seule façon de le faire? Il doit sûrement y avoir un moyen de lire à partir de plusieurs flux sans utiliser un fil pour chacun?

Answer 1

Comme vous l'avez dit, la solution décrite dans cette Answer est le moyen traditionnel de lire stdout et stderr à partir d'un processus. Un thread par flux est la voie à suivre, même si cela est légèrement gênant.

Answer 2

Vous devrez en effet choisir de générer un thread pour chaque flux que vous souhaitez surveiller. Si votre cas d'utilisation permet de combiner stdout et stderr du processus en question, vous n'avez besoin que d'un seul thread, sinon deux sont nécessaires.

Il m'a fallu un certain temps pour bien faire les choses dans l'un de nos projets où je devais lancer un processus externe, en extraire le résultat et en faire quelque chose tout en recherchant des erreurs, en mettant fin au processus et en étant capable pour la terminer lorsque l'utilisateur de l'application java annule l'opération.

J'ai créé une classe assez simple pour encapsuler la partie regardante dont la méthode run () ressemble à ceci:

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

Du côté des appels, cela ressemble à ceci:

    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;

Malheureusement, je n'ai pas à le faire pour un exemple autonome exécutable, mais peut-être que cela aide. Si je devais le refaire, j'examinerais de nouveau l'utilisation de la méthode Thread.interrupt () au lieu d'un indicateur d'arrêt auto-fabriqué (pensez à le déclarer volatile!), Mais je le laisse pour une autre fois. :)