Java -多并发runtime.exec() InputStreams的问题

Question

我别无选择，只能通过几次对Runtime.exec()的VBScript调用来检索一些外部数据。我真的很讨厌这个实现，因为我失去了跨平台的灵活性，但我最终可能会开发类似的*nix脚本，至少可以减轻这个问题。在有人问我之前，我无法绕过调用外部脚本来收集数据的需要。我会忍受所造成的问题。

exec()进程在扩展Runnable的自定义类中运行。它使用BufferedReader从getInputStream()读取数据。

编辑__：更多的代码是根据请求添加的，但是我看不出额外的代码有什么关系:)我希望它有帮助，因为格式化需要一段时间！哦，如果我的代码风格很难看的话，那就放轻松点，但是我们鼓励建设性的批评.

public class X extends JFrame implements Runnable {

   ...
   static final int THREADS_MAX = 4;
   ExecutorService  exec;
   ...
   public static void main(String[] args) {
      ...
      SwingUtilities.invokeLater(new X("X"));
   } // End main(String[])

   public X (String title) {
      ...
      exec = Executors.newFixedThreadPool(THREADS_MAX);
      ...

      // Create all needed instances of Y
      for (int i = 0; i < objects.length; i++) {
         Y[i] = new Y(i);
      } // End for(i)

      // Initialization moved here for easy single-thread testing
      // Undesired, of course
      for (int i = 0; i < objects.length; i++) {
         Y[i].initialize(parent);
      } // End for(i)

   } // End X

   class Y implements Runnable {
      // Define variables/arrays used to capture data here
      String computerName = "";
      ...

      public Y(int rowIndex) {
         row          = rowIndex;
         ...
         computerName = (String)JTable.getValueAt(row, 0);
         ...
         exec.execute(this);
      } // End Y(int)

      public void run() {
         // Initialize variables/arrays used to capture data here
         ...

         // Initialization should be done here for proper threading
         //initialize(parent);
      } // End run()

      public void initialize(Z obj) {
         runTime = Runtime.getRuntime();
         ...

         try {
            process = runTime.exec("cscript.exe query.vbs " + computerName);
            stdErr  = process.getErrorStream();
            stdIn   = process.getInputStream();
            isrErr  = new InputStreamReader(stdErr);
            isrIn   = new InputStreamReader(stdIn);
            brErr   = new BufferedReader(isrErr);
            brIn    = new BufferedReader(isrIn);

            while ((line = brIn.readLine()) != null) {
               // Capture, parse, and store data here
               ...
            } // End while

         } catch (IOException e) {
            System.out.println("Unable to run script");
         } catch (Exception e) {
            e.printStackTrace();
         } finally {
            try {
               stdErr.close();
               stdIn. close();
               isrErr.close();
               isrIn. close();
               brErr. close();
               brIn.  close();
            } catch (IOException e) {
               System.out.println("Unable to close streams.");
            } // End try
         } // End try
      } // End initialize(Z)
      ...
   } // End class Y
} // End class X

如果我单独执行这些命令，我就会像我期望的那样收集数据。但是，如果我在类的run()块中执行命令(这意味着调用是并发的，正如我所希望的那样)，似乎只生成了一个输入流，所有BufferedReaders都同时使用该流。

为了调试这个问题，我在控制台上以类的实例读取输入流为前缀输出每一行。我期望类似这样的东西，理解它们在实例到实例之间可能是无序的，但是单个实例的行顺序是完整的：

exec 0: Line1
exec 1: Line1
exec 2: Line1
exec 0: Line2
exec 1: Line2
exec 2: Line2
exec 0: Line3
exec 1: Line3
exec 2: Line3
...

奇怪的是，我得到了输出的第一行(Microsoft (R) Windows Script Host Version 5.7)的预期实例数，但是在这一行之后，只有一个进程继续在输入流中生成数据，所有读者都随机使用这一个流，如下面的示例：

exec 2: Microsoft (R) Windows Script Host Version 5.7
exec 0: Microsoft (R) Windows Script Host Version 5.7
exec 1: Microsoft (R) Windows Script Host Version 5.7
exec 0: line2
exec 1: line3
exec 2: line4
...

更糟糕的是，读取器会停止，readLine()永远不会返回null。我读到这种类型的行为可能与缓冲区大小有关，但是当我只运行两个并发线程时，即使输出很短，它仍然表现出相同的行为。在stdErr中没有捕获任何信息来表示存在问题。

为了查看这是否是脚本主机的限制，我创建了一个批处理文件，该文件是START并发的多个脚本实例。I应该声明这是在的外部运行的，在一个cmd中，并启动它自己的几个shell。然而，每个并发实例都完全返回预期的结果，并且运行良好。

编辑：作为另一个疑难解答的想法，我决定重新启用并发，但通过将以下内容插入到Y.run()块中来错开初始化方法：

try {
   Thread.sleep((int)(Math.random() * 1200));
} catch (InterruptedException e) {
   System.out.println("Can't sleep!");
} // End try
initialize(monitor);

进入我的密码。我开始看到前几行的多个输出，但它很快会恢复到多个使用同一个生产者的消费者，一旦第一个完成的流关闭，其余的使用者就会触发异常。下一个用户会触发一个IOException: Read error，其余的会触发IOException: Stream closed！

根据maaartinus的说法，可以运行多个并发的InputStreams，所以现在的问题是是什么导致了不想要的行为？我如何独立地获取它们的输入流？如果我能避免的话，我不想仅仅为了处理返回的数据而将其写入临时文件。

Answer 1

我认为您需要小心IO变量的范围。下面是一个工作非常好的快速代码，它使用来自4个子进程的并发输入流.

import java.io.*;

public class MultiExec {

        private final static String[] comLines = {
                        "date",
                        "ls /var/spool/postfix",
                        "ls -F /usr/local/bin",
                        "wc -l /etc/apache2/apache2.conf"};

        public void execute() {
                for (int i = 0 ; i < comLines.length ; i++) {
                        ExecutableChild ec = new ExecutableChild (i, comLines[i]);
                        new Thread (ec).start();
        }}

        public class ExecutableChild implements Runnable {

                private int prIndex;
                private String executable;

                public ExecutableChild (int k, String cmd) {
                        prIndex = k;
                        executable = cmd;
                }

                public void run () {
                        try {
                                Process child = Runtime.getRuntime().exec(executable);
                                BufferedReader br = new BufferedReader (new InputStreamReader (
                                                                child.getInputStream()));
                                for (String s = br.readLine() ; s != null ; s = br.readLine())
                                        System.out.println ("[" + prIndex + "] " + s);
                                br.close();
                        } catch (IOException ioex) {
                                System.err.println ("IOException for process #"+
                                                prIndex+ ": " + ioex.getMessage());
        }}}

        public static void main (String[] args) {
                new MultiExec().execute();
        }
}

来自上述代码的输出(% javac MultiExec.java；java MultiExec)

[2] tomcat*
[0] Thu Jan 20 18:38:31 CST 2011
[3] 368 /etc/apache2/apache2.conf
[1] active
[1] bounce
[1] corrupt
[1] defer
[1] deferred
[1] etc
[1] flush
[1] hold
[1] incoming
[1] lib
[1] maildrop
[1] pid
[1] private
[1] public
[1] saved
[1] trace
[1] usr
[1] var

如果你给我们你的尝试的源代码，我们可以讨论它。祝你，- M.S.

=============================================================================

编辑: DN:我理解你对1行输出的关注.我们有个小剧本..。

#!/usr/bin/perl -w
foreach (1..50) {
        print "$_\n";
}

以及上述Java代码的编辑版本..。comLines已经更改，并且在每个println()之后添加了一个Thread.sleep

公共类MultiExec {

        private final static String[] comLines = {
                        "ls /var/spool/postfix",
                        "perl count50.pl",
                        "cat MultiExec.java",
                        "head -40 /etc/apache2/apache2.conf"};

        public void execute() {
                for (int i = 0 ; i < comLines.length ; i++) {
                        ExecutableChild ec = new ExecutableChild (i, comLines[i]);
                        new Thread (ec).start();
        }}

        public class ExecutableChild implements Runnable {

                private int prIndex;
                private String executable;

                public ExecutableChild (int k, String cmd) {
                        prIndex = k;
                        executable = cmd;
                }

                public void run () {
                        try {
                                Process child = Runtime.getRuntime().exec(executable);
                                BufferedReader br = new BufferedReader (new InputStreamReader (
                                                                child.getInputStream()));
                                for (String s = br.readLine() ; s != null ; s = br.readLine()) {
                                        System.out.println ("[" + prIndex + "] " + s);
                                        try {
                                                Thread.sleep (20);
                                        } catch (InterruptedException intex) {
                                }}
                                br.close();
                        } catch (IOException ioex) {
                                System.err.println ("IOException for process #"+
                                                                prIndex+ ": " + ioex.getMessage());
        }}}

        public static void main (String[] args) {
                new MultiExec().execute();
}}

下面是输出(编译/运行后).

[0] active
[1] 1
[2] import java.io.*;
[3] #
[2]
[0] bounce
[1] 2
[3] # Based upon the NCSA server configuration files originally by Rob McCool.
[2] public class MultiExec {
[1] 3
[0] corrupt
[3] #
[1] 4
[2]
[0] defer
[3] # This is the main Apache server configuration file.  It contains the
[2]     private final static String[] comLines = {
[0] deferred
[1] 5
[3] # configuration directives that give the server its instructions.
[2]                     "ls /var/spool/postfix",
[0] etc
[1] 6
[3] # See http://httpd.apache.org/docs/2.2/ for detailed information about
[2]                     "perl count50.pl",
[0] flush
[1] 7
[3] # the directives.
[2]                     "cat MultiExec.java",
[1] 8
[0] hold
[3] #
[1] 9
[2]                     "head -40 /etc/apache2/apache2.conf"};
[0] incoming
[3] # Do NOT simply read the instructions in here without understanding
[2]
[0] lib
[1] 10
[3] # what they do.  They're here only as hints or reminders.  If you are unsure
[1] 11
[2]     public void execute() {
[0] maildrop
[3] # consult the online docs. You have been warned.
[2]             for (int i = 0 ; i < comLines.length ; i++) {
[0] pid
[1] 12
[3] #
[1] 13
[2]                     ExecutableChild ec = new ExecutableChild (i, comLines[i]);
[0] private
[3] # The configuration directives are grouped into three basic sections:
[1] 14
[2]                     new Thread (ec).start();
[0] public
[3] #  1. Directives that control the operation of the Apache server process as a
[2]     }}
[1] 15
[0] saved
[3] #     whole (the 'global environment').
[1] 16
[0] trace
[2]
[3] #  2. Directives that define the parameters of the 'main' or 'default' server,
[0] usr
[2]     public class ExecutableChild implements Runnable {
[1] 17
[3] #     which responds to requests that aren't handled by a virtual host.
[0] var
[2]
[1] 18
[3] #     These directives also provide default values for the settings
[1] 19
[2]             private int prIndex;
[3] #     of all virtual hosts.
[1] 20
[2]             private String executable;
[3] #  3. Settings for virtual hosts, which allow Web requests to be sent to
[2]
[1] 21
[3] #     different IP addresses or hostnames and have them handled by the
[1] 22
[2]             public ExecutableChild (int k, String cmd) {
[3] #     same Apache server process.
[1] 23
[2]                     prIndex = k;
[3] #
[1] 24
[2]                     executable = cmd;
[3] # Configuration and logfile names: If the filenames you specify for many
[2]             }
[1] 25
[3] # of the server's control files begin with "/" (or "drive:/" for Win32), the
[2]
[1] 26
[3] # server will use that explicit path.  If the filenames do *not* begin
[1] 27
[2]             public void run () {
[3] # with "/", the value of ServerRoot is prepended -- so "/var/log/apache2/foo.log"
[1] 28
[2]                     try {
[3] # with ServerRoot set to "" will be interpreted by the
[1] 29
[2]                             Process child = Runtime.getRuntime().exec(executable);
[3] # server as "//var/log/apache2/foo.log".
[1] 30
[2]                             BufferedReader br = new BufferedReader (new InputStreamReader (
[3] #
[1] 31
[2]                                                             child.getInputStream()));
[3]
[1] 32
[2]                             for (String s = br.readLine() ; s != null ; s = br.readLine()) {
[3] ### Section 1: Global Environment
[1] 33
[2]                                     System.out.println ("[" + prIndex + "] " + s);
[3] #
[1] 34
[2]                                     try {
[3] # The directives in this section affect the overall operation of Apache,
[1] 35
[2]                                             Thread.sleep (20);
[3] # such as the number of concurrent requests it can handle or where it

......

输入流工作得很好，不要以为我在这里有问题。很抱歉回复时间太长了。祝你一切顺利，等着看你的代码，- M.S。

Answer 2

确保在正确的范围内声明了stdErr和stdIn。在这种情况下，您需要在Y中声明它们。

如果要在X中声明它们，则每次运行以下代码：

stdErr  = process.getErrorStream();
stdIn   = process.getInputStream();

变量将重新分配，所有Y实例都将引用相同的流。