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

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 等)可能会以不同的方式处理“刷新”操作，也许您最终收到的是该外部进程中任何给定刷新的多个字节 block 。

在任何情况下，您都可以尝试使用 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)。