是否可以从超时的输入流中读取数据?

2022-08-31 07:09:19

具体来说,问题是要写一个这样的方法:

int maybeRead(InputStream in, long timeout)

其中,如果数据在“超时”毫秒内可用,则返回值与 in.read() 相同,否则返回值为 -2。在该方法返回之前,任何生成的线程都必须退出。

为了避免争论,这里的主题是java.io.InputStream,正如Sun(任何Java版本)所记录的那样。请注意,这并不像看起来那么简单。以下是Sun文档直接支持的一些事实。

  1. in.read() 方法可能是不可中断的。

  2. 将 InputStream 包装在读取器或可中断通道中没有帮助,因为所有这些类都可以调用 InputStream 的方法。如果可以使用这些类,则可以编写一个解决方案,该解决方案仅直接在 InputStream 上执行相同的逻辑。

  3. in.available() 返回 0 始终是可以接受的。

  4. in.close() 方法可能会阻止或不执行任何操作。

  5. 没有一般的方法来杀死另一个线程。


答案 1

使用 inputStream.available()

System.in.available() 返回 0 始终是可以接受的。

我发现相反 - 它总是返回可用字节数的最佳值。Javadoc for :InputStream.available()

Returns an estimate of the number of bytes that can be read (or skipped over) 
from this input stream without blocking by the next invocation of a method for 
this input stream.

由于时间/过时性,估计是不可避免的。这个数字可能是一次性低估,因为新的数据不断到来。但是,它总是在下一个呼叫时“赶上” - 它应该考虑所有到达的数据,除非在新呼叫的那一刻到达。当有数据未通过上述条件时,永久返回 0。

第一个警告:InputStream的具体子类负责可用()

InputStream是一个抽象类。它没有数据源。对于它来说,拥有可用的数据是没有意义的。因此,javadoc for 也声明:available()

The available method for class InputStream always returns 0.

This method should be overridden by subclasses.

事实上,具体的输入流类确实覆盖了 available(),提供了有意义的值,而不是常量 0。

第二个警告:确保在 Windows 中键入输入时使用回车符。

如果使用 ,则仅当命令 shell 交出输入时,程序才会接收输入。如果您使用的是文件重定向/管道(例如,>java myJavaApp或java myJavaApp|某些命令),则输入数据通常会立即移交。但是,如果手动键入输入,则数据切换可能会延迟。例如,使用windows cmd.exe shell,数据在cmd.exe shell中缓冲。数据仅在回车(control-m 或 )之后传递到正在执行的 java 程序。这是执行环境的局限性。当然,InputStream.available() 将返回 0,只要 shell 缓冲数据 - 这是正确的行为;此时没有可用的数据。一旦数据从 shell 中可用,该方法将返回一个值 > 0。注意:Cygwin也使用cmd.exe。System.in<enter>

最简单的解决方案(无阻塞,因此无需超时)

只需使用这个:

    byte[] inputData = new byte[1024];
    int result = is.read(inputData, 0, is.available());  
    // result will indicate number of bytes read; -1 for EOF with no data read.

或者等价地,

    BufferedReader br = new BufferedReader(new InputStreamReader(System.in, Charset.forName("ISO-8859-1")),1024);
    // ...
         // inside some iteration / processing logic:
         if (br.ready()) {
             int readCount = br.read(inputData, bufferOffset, inputData.length-bufferOffset);
         }

更丰富的解决方案(在超时期限内最大限度地填充缓冲区)

声明如下:

public static int readInputStreamWithTimeout(InputStream is, byte[] b, int timeoutMillis)
     throws IOException  {
     int bufferOffset = 0;
     long maxTimeMillis = System.currentTimeMillis() + timeoutMillis;
     while (System.currentTimeMillis() < maxTimeMillis && bufferOffset < b.length) {
         int readLength = java.lang.Math.min(is.available(),b.length-bufferOffset);
         // can alternatively use bufferedReader, guarded by isReady():
         int readResult = is.read(b, bufferOffset, readLength);
         if (readResult == -1) break;
         bufferOffset += readResult;
     }
     return bufferOffset;
 }

然后使用这个:

    byte[] inputData = new byte[1024];
    int readCount = readInputStreamWithTimeout(System.in, inputData, 6000);  // 6 second timeout
    // readCount will indicate number of bytes read; -1 for EOF with no data read.

答案 2

假设你的流没有套接字支持(所以你不能使用),我认为解决这类问题的标准方法是使用Future。Socket.setSoTimeout()

假设我有以下执行器和流:

    ExecutorService executor = Executors.newFixedThreadPool(2);
    final PipedOutputStream outputStream = new PipedOutputStream();
    final PipedInputStream inputStream = new PipedInputStream(outputStream);

我有一个编写器,它写入一些数据,然后等待5秒钟,然后再写入最后一段数据并关闭流:

    Runnable writeTask = new Runnable() {
        @Override
        public void run() {
            try {
                outputStream.write(1);
                outputStream.write(2);
                Thread.sleep(5000);
                outputStream.write(3);
                outputStream.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    };
    executor.submit(writeTask);

阅读本文的正常方式如下。读取将无限期地阻止数据,因此这将在5秒内完成:

    long start = currentTimeMillis();
    int readByte = 1;
    // Read data without timeout
    while (readByte >= 0) {
        readByte = inputStream.read();
        if (readByte >= 0)
            System.out.println("Read: " + readByte);
    }
    System.out.println("Complete in " + (currentTimeMillis() - start) + "ms");

其中输出:

Read: 1
Read: 2
Read: 3
Complete in 5001ms

如果有一个更根本的问题,比如作者没有回应,读者将永远阻止。如果我将来将读取包装在一起,则可以按如下方式控制超时:

    int readByte = 1;
    // Read data with timeout
    Callable<Integer> readTask = new Callable<Integer>() {
        @Override
        public Integer call() throws Exception {
            return inputStream.read();
        }
    };
    while (readByte >= 0) {
        Future<Integer> future = executor.submit(readTask);
        readByte = future.get(1000, TimeUnit.MILLISECONDS);
        if (readByte >= 0)
            System.out.println("Read: " + readByte);
    }

其中输出:

Read: 1
Read: 2
Exception in thread "main" java.util.concurrent.TimeoutException
    at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:228)
    at java.util.concurrent.FutureTask.get(FutureTask.java:91)
    at test.InputStreamWithTimeoutTest.main(InputStreamWithTimeoutTest.java:74)

我可以捕获TimeoutException并执行我想要的任何清理。


推荐