具有并发输入/输出流的Java流程

陶高扬

2023-03-14

问题内容：

我正在尝试创建一种控制台/终端，允许用户输入一个字符串，然后将其编入进程并打印出结果。就像普通的控制台一样。但是我在管理输入/输出流时遇到了麻烦。我已经研究了这个线程，但是可悲的是，该解决方案不适用于我的问题。

与标准命令（例如“ ipconfig”和“ cmd.exe”）一起，如果脚本要求输入，我还需要能够运行脚本并使用相同的输入流传递一些参数。

例如，在运行脚本“ python
pyScript.py”之后，如果它要求的话，我应该能够将进一步的输入传递给脚本（例如：raw_input），同时还要打印脚本的输出。您期望从终端获得的基本行为。

到目前为止，我得到的是：

import java.awt.BorderLayout;
import java.awt.Color;
import java.awt.Dimension;
import java.awt.event.KeyEvent;
import java.awt.event.KeyListener;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;

import javax.swing.JFrame;
import javax.swing.JPanel;
import javax.swing.JScrollPane;
import javax.swing.JTextPane;
import javax.swing.text.BadLocationException;
import javax.swing.text.Document;

public class Console extends JFrame{

    JTextPane inPane, outPane;
    InputStream inStream, inErrStream;
    OutputStream outStream;

    public Console(){
        super("Console");
        setPreferredSize(new Dimension(500, 600));
        setLocationByPlatform(true);
        setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);

        // GUI
        outPane = new JTextPane();
        outPane.setEditable(false);
        outPane.setBackground(new Color(20, 20, 20));
        outPane.setForeground(Color.white);
        inPane = new JTextPane();
        inPane.setBackground(new Color(40, 40, 40));
        inPane.setForeground(Color.white);
        inPane.setCaretColor(Color.white);

        JPanel panel = new JPanel(new BorderLayout());
        panel.add(outPane, BorderLayout.CENTER);
        panel.add(inPane, BorderLayout.SOUTH);

        JScrollPane scrollPanel = new JScrollPane(panel);

        getContentPane().add(scrollPanel);

        // LISTENER
        inPane.addKeyListener(new KeyListener(){
            @Override
            public void keyPressed(KeyEvent e){
              if(e.getKeyCode() == KeyEvent.VK_ENTER){
                    e.consume();
                    read(inPane.getText());
                }
            }
            @Override
            public void keyTyped(KeyEvent e) {}

            @Override
            public void keyReleased(KeyEvent e) {}
        });


        pack();
        setVisible(true);
    }

    private void read(String command){
        println(command);

        // Write to Process
        if (outStream != null) {
            System.out.println("Outstream again");
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(outStream));
            try {
                writer.write(command);
                //writer.flush();
                //writer.close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }

        // Execute Command
        try {
            exec(command);
        } catch (IOException e) {}

        inPane.setText("");
    }

    private void exec(String command) throws IOException{
        Process pro = Runtime.getRuntime().exec(command, null);

        inStream = pro.getInputStream();
        inErrStream = pro.getErrorStream();
        outStream = pro.getOutputStream();

        Thread t1 = new Thread(new Runnable() {
            public void run() {
                try {
                    String line = null;
                    while(true){
                        BufferedReader in = new BufferedReader(new InputStreamReader(inStream));
                        while ((line = in.readLine()) != null) {
                            println(line);
                        }
                        BufferedReader inErr = new BufferedReader(new InputStreamReader(inErrStream));
                        while ((line = inErr.readLine()) != null) {
                            println(line);
                        }
                        Thread.sleep(1000);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        t1.start();
    }

    public void println(String line) {
        Document doc = outPane.getDocument();
        try {
            doc.insertString(doc.getLength(), line + "\n", null);
        } catch (BadLocationException e) {}
    }

    public static void main(String[] args){
        new Console();
    }
}

我不使用提及的内容ProcessBuilder，因为我喜欢区分错误流和正常流。

更新29.08.2016

在@ArcticLord的帮助下，我们实现了原始问题中的要求。现在，只需解决所有奇怪的行为，例如非终止过程。控制台具有一个“停止”按钮，该按钮仅调用pro.destroy（）。但是由于某种原因，这对于无限运行的进程是不起作用的，这些进程正在向垃圾邮件发送。

控制台：http :
//pastebin.com/vyxfPEXC

InputStreamLineBuffer：http：//pastebin.com/TzFamwZ1

不会停止的示例代码：

public class Infinity{
    public static void main(String[] args){ 
        while(true){
            System.out.println(".");
        }
    }
}

停止的示例代码：

import java.util.concurrent.TimeUnit;

public class InfinitySlow{
    public static void main(String[] args){ 
        while(true){
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(".");
        }
    }
}

问题答案：

您的代码使用正确的方法。您只错过了一些小事。
让我们从您的read方法开始：

private void read(String command){
    [...]
    // Write to Process
    if (outStream != null) {
        [...]
        try {
            writer.write(command + "\n");  // add newline so your input will get proceed
            writer.flush();  // flush your input to your process
        } catch (IOException e1) {
            e1.printStackTrace();
        }
    }
    // ELSE!! - if no outputstream is available
    // Execute Command
    else {
        try {
            exec(command);
        } catch (IOException e) {
            // Handle the exception here. Mostly this means
            // that the command could not get executed
            // because command was not found.
            println("Command not found: " + command);
        }
    }
    inPane.setText("");
}

现在让我们修复您的exec方法。您应该使用单独的线程来读取常规进程输出和错误输出。另外，我引入了第三个线程，该线程等待进程结束并关闭outputStream，因此下一个用户输入不用于进程，而是一个新命令。

private void exec(String command) throws IOException{
    Process pro = Runtime.getRuntime().exec(command, null);

    inStream = pro.getInputStream();
    inErrStream = pro.getErrorStream();
    outStream = pro.getOutputStream();

    // Thread that reads process output
    Thread outStreamReader = new Thread(new Runnable() {
        public void run() {
            try {
                String line = null;
                BufferedReader in = new BufferedReader(new InputStreamReader(inStream));                        
                while ((line = in.readLine()) != null) {
                    println(line);                       
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("Exit reading process output");
        }
    });
    outStreamReader.start();

    // Thread that reads process error output
    Thread errStreamReader = new Thread(new Runnable() {
        public void run() {
            try {
                String line = null;           
                BufferedReader inErr = new BufferedReader(new InputStreamReader(inErrStream));
                while ((line = inErr.readLine()) != null) {
                    println(line);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("Exit reading error stream");
        }
    });
    errStreamReader.start();

    // Thread that waits for process to end
    Thread exitWaiter = new Thread(new Runnable() {
        public void run() {
            try {
                int retValue = pro.waitFor();
                println("Command exit with return value " + retValue);
                // close outStream
                outStream.close();
                outStream = null;
            } catch (InterruptedException e) {
                e.printStackTrace(); 
            } catch (IOException e) {
                e.printStackTrace();
            } 
        }
    });
    exitWaiter.start();
}

现在这应该工作。
如果输入，ipconfig它将打印命令输出，关闭输出流并准备好使用新命令。
如果输入，cmd它将打印输出，并允许您输入更多cmd命令，例如diror
cd等等，直到输入exit。然后，它关闭输出流，并准备好执行新命令。

您可能会在执行python脚本时遇到问题，因为如果未将它们刷新到系统管道中，则使用Java读取Process InputStream会出现问题。
参见此示例python脚本

print "Input something!"
str = raw_input()
print "Received input is : ", str

您可以使用Java程序来运行它，也可以输入输入，但是直到脚本完成后您才能看到脚本输出。
我能找到的唯一解决方法是手动刷新脚本中的输出。

import sys
print "Input something!"
sys.stdout.flush()
str = raw_input()
print "Received input is : ", str
sys.stdout.flush()

运行此脚本将按照您的预期进行。

编辑：我刚刚找到了另一个非常简单的解决stdout.flush()Python脚本问题的方法。使用它们启动python -u script.py，您无需手动冲洗。这应该可以解决您的问题。

EDIT2：我们在评论中讨论过，由于此解决方案的输出和错误流在不同的线程中运行，因此它们会混合在一起。这里的问题是，当出现错误流线程时，我们无法区分输出写入是否完成。否则，带锁的经典线程调度可以处理这种情况。但是无论数据是否流动，我们都有一个连续的流，直到处理完成。因此，我们需要一种机制来记录自从每个流中读取最后一行以来经过了多少时间。

为此，我将介绍一个类，该类获取InputStream并启动一个Thread来读取传入的数据。该线程将每行存储在队列中，并在流的末尾到达时停止。另外，它保留读取最后一行并将其添加到队列的时间。

public class InputStreamLineBuffer{
    private InputStream inputStream;
    private ConcurrentLinkedQueue<String> lines;
    private long lastTimeModified;
    private Thread inputCatcher;
    private boolean isAlive;

    public InputStreamLineBuffer(InputStream is){
        inputStream = is;
        lines = new ConcurrentLinkedQueue<String>();
        lastTimeModified = System.currentTimeMillis();
        isAlive = false;
        inputCatcher = new Thread(new Runnable(){
            @Override
            public void run() {
                StringBuilder sb = new StringBuilder(100);
                int b;
                try{
                    while ((b = inputStream.read()) != -1){  
                        // read one char
                        if((char)b == '\n'){
                            // new Line -> add to queue
                            lines.offer(sb.toString());
                            sb.setLength(0); // reset StringBuilder
                            lastTimeModified = System.currentTimeMillis();
                        }
                        else sb.append((char)b); // append char to stringbuilder
                    }
                } catch (IOException e){
                    e.printStackTrace();
                } finally {
                    isAlive = false;
                }
            }});
    }
    // is the input reader thread alive
    public boolean isAlive(){
        return isAlive;
    }
    // start the input reader thread
    public void start(){
        isAlive = true;
        inputCatcher.start();
    }
    // has Queue some lines
    public boolean hasNext(){
        return lines.size() > 0;
    }
    // get next line from Queue
    public String getNext(){
        return lines.poll();
    }
    // how much time has elapsed since last line was read
    public long timeElapsed(){
        return (System.currentTimeMillis() - lastTimeModified);
    }
}

使用此类，我们可以将输出和错误读取线程合并为一个。当输入读取缓冲区线程处于活动状态并且没有消耗数据时，这种状态仍然存在。在每次运行中，它都会检查自从读取最后一个输出以来是否已经过了一段时间，如果是，它将一口气打印所有未打印的行。与错误输出相同。然后因为不浪费CPU时间而睡了几毫秒。

private void exec(String command) throws IOException{
    Process pro = Runtime.getRuntime().exec(command, null);

    inStream = pro.getInputStream();
    inErrStream = pro.getErrorStream();
    outStream = pro.getOutputStream();

    InputStreamLineBuffer outBuff = new InputStreamLineBuffer(inStream);
    InputStreamLineBuffer errBuff = new InputStreamLineBuffer(inErrStream);

    Thread streamReader = new Thread(new Runnable() {       
        public void run() {
            // start the input reader buffer threads
            outBuff.start();
            errBuff.start();

            // while an input reader buffer thread is alive
            // or there are unconsumed data left
            while(outBuff.isAlive() || outBuff.hasNext() ||
                errBuff.isAlive() || errBuff.hasNext()){

                // get the normal output if at least 50 millis have passed
                if(outBuff.timeElapsed() > 50)
                    while(outBuff.hasNext())
                        println(outBuff.getNext());
                // get the error output if at least 50 millis have passed
                if(errBuff.timeElapsed() > 50)
                    while(errBuff.hasNext())
                        println(errBuff.getNext());
                // sleep a bit bofore next run
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }                 
            }
            System.out.println("Finish reading error and output stream");
        }          
    });
    streamReader.start();

    // remove outStreamReader and errStreamReader Thread
    [...]
}

也许这不是一个完美的解决方案，但是应该可以解决这里的情况。

编辑（31.8.2016）
我们在注释中讨论了在实现停止按钮以终止使用停止已启动的过程时，代码仍然存在问题Process#destroy()。产生大量输出（例如在无限循环中）的进程将通过调用立即销毁destroy()。但是，由于它已经产生了许多必须由我们使用的输出，因此streamReader我们无法恢复正常的程序行为。
因此，我们需要在此处进行一些小的更改：

我们将向引入一个destroy()方法，该方法将InputStreamLineBuffer停止输出读取并清除队列。
更改将如下所示：

public class InputStreamLineBuffer{
    private boolean emergencyBrake = false;
    [...]
    public InputStreamLineBuffer(InputStream is){
        [...]
                while ((b = inputStream.read()) != -1 && !emergencyBrake){
                    [...]
                }
    }
    [...]

    // exits immediately and clears line buffer
    public void destroy(){
        emergencyBrake = true;
        lines.clear();
    }
}

还有主程序中的一些小变化

public class ExeConsole extends JFrame{
    [...]
    // The line buffers must be declared outside the method
    InputStreamLineBuffer outBuff, errBuff; 
    public ExeConsole{
        [...]
        btnStop.addActionListener(new ActionListener() {
            public void actionPerformed(ActionEvent e) {
                 if(pro != null){
                      pro.destroy();
                      outBuff.destroy();
                      errBuff.destroy();
                 }
        }});
    }
    [...]
    private void exec(String command) throws IOException{
        [...]
        //InputStreamLineBuffer outBuff = new InputStreamLineBuffer(inStream);
        //InputStreamLineBuffer errBuff = new InputStreamLineBuffer(inErrStream);        
        outBuff = new InputStreamLineBuffer(inStream);
        errBuff = new InputStreamLineBuffer(inErrStream);    
        [...]
    }
}

现在，它应该甚至可以销毁某些输出垃圾邮件的过程。

注意：
我发现这Process#destroy()无法破坏子进程。因此，如果您cmd在Windows上启动并从那里启动Java程序，则最终将cmd在Java程序仍在运行时破坏该进程。您将在任务管理器中看到它。Java本身无法解决此问题。它需要一些依赖于外部工具的操作系统来获取这些进程的pid，并手动将其杀死。

具有并发输入/输出流的Java流程

相关阅读

相关文章

相关问答

相关工具

相关文档