如何在不阻塞的情况下将进程的标准输出作为流使用？

Question

在 Java（或 clojure）中，我想启动一个外部进程并将其标准输出作为流使用。理想情况下，我希望每次外部进程刷新它时都使用该进程的输出流，但不确定如何实现，以及如何在不阻塞的情况下实现。

四处使用 Java ProcessPipeInputStream 进行脱壳进程（例如 a Unix ProcessPipeInputStream），我发现继承的 InputStream 方法使用起来有点低级，并且不确定是否存在非每次生产者端刷新时从流中使用的阻塞方式或以非阻塞方式。

许多代码示例会在无限循环中阻塞输出流，从而占用一个线程进行监听。我希望这种阻塞行为可以完全避免。

底线：

是否有一种非阻塞方式来通知输入流，每次它的生产端刷新时？

Answer 1

您需要创建一个单独的线程，该线程将从这样的流中消耗，从而允许您的程序的其余部分并行执行本应执行的操作。

class ProcessOutputReader implements Runnable {

   private InputStream processOutput;

   public ProcessOutputReader(final InputStream processOutput) {
       this.processOutput = processOutput;
   } 

   @Override
   public void run() {
      int nextByte;
      while ((nextByte = processOutput.read()) != -1) {
        // do whatever you need to do byte-by-byte.
        processByte(nextByte);
      }
   }
}

class Main {
   public static void main(final String[] args) {
       final Process proc = ...; 
       final ProcessOutputReader reader = new ProcessOutputReader(proc.getInputStream());
       final Thread processOutputReaderThread = new Thread(reader);
       processOutputReaderThread.setDaemon(true); // allow the VM to terminate if this is the only thread still active.
       processOutputReaderThread.start();
       ...
       // if you wanna wait for the whole process output to be processed at some point you can do this:
       try {
         processOutputReaderThread.join();
       } catch (final InterruptedException ex) {
         // you need to decide how to recover from if your wait was interrupted.
       }
   }
}

如果您不想逐字节处理，而是希望将每个刷新作为一个单独的部分来处理...我不确定是否 100% 保证能够捕获每个进程刷新。毕竟进程自己的 IO 框架软件（Java、C、Python 等）可能会以不同的方式处理“刷新”操作，并且您最终收到的可能是该外部进程中任何给定刷新的多个字节块。

在任何情况下，您都可以尝试使用InputStream 的available 方法，如下所示：

   @Override
   public void run() {
      int nextByte;
      while ((nextByte = processOutput.read()) != -1) {
        final int available = processOutput.available();
        byte[] block = new byte[available + 1];
        block[0] = nextByte;
        final int actuallyAvailable = processOutput.read(block, 1, available);
        if (actuallyAvailable < available) { 
          if (actuallyAvailable == -1) { 
            block = new byte[] { nextByte };
          } else {
            block = Arrays.copyOf(block, actuallyAvailable + 1);
          } 
        }
        // do whatever you need to do on that block now.
        processBlock(block);
      }
   }

我不能 100% 确定这一点，但我认为不能相信 available 将返回一个保证的字节数下限，您可以在不被阻塞的情况下检索，也不能相信下一个 read 操作是如果要求，将返回该数量的 可用 字节；这就是为什么上面的代码检查实际读取的字节数 (actuallyAvailable)。