Java：合并输入流

Question

我的目标是创建（或使用现有的）一个 InputStream 实现（例如 MergeInputStream），它将尝试从多个 InputStream 中读取并返回第一个结果。之后，它将释放锁定并停止从所有 InputStreams 读取，直到下一次 mergeInputStream.read() 调用。我很惊讶我没有找到任何这样的工具。问题是：所有源 InputStreams 都不是很有限（不是文件，例如，而是 System.in、socket 等），所以我不能使用 SequenceInputReader。我知道这可能需要一些多线程机制，但我完全不知道该怎么做。我尝试谷歌搜索但没有结果。

Answer 1

最好使用SelectableChannel和Selector解决从多个源读取输入并将它们序列化为一个流的问题。然而，这要求所有源都能够提供可选择的频道。这可能是也可能不是。

如果可选通道不可用，您可以选择使用单个线程解决它，让读取实现执行以下操作：对于每个输入流is，检查is.available() > 0 ，如果是，则返回is.read()。重复此过程，直到某些输入流有可用数据。

但是，这种方法有两个主要缺点：

1. NotallimplementationsofInputStream 以某种方式实现available()，当且仅当read() 将阻塞时它才返回0。结果自然是无法从此流中读取数据，即使is.read() 会返回一个值。这是否被认为是一个错误是值得怀疑的，因为文档只是声明它应该返回一个可用字节数的“估计”。

2. 它使用所谓的“忙循环”，这基本上意味着您需要在循环中设置睡眠（这会导致读取延迟）或不必要地占用 CPU。

您的第三种选择是通过为每个输入流生成一个线程来处理阻塞读取。但是，如果要读取的输入流数量非常多，这将需要仔细同步，并且可能需要一些开销。下面的代码是第一次尝试解决它。我不确定它是否充分同步，或者它是否以最佳方式管理线程。

import java.io.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class MergedInputStream extends InputStream {

    AtomicInteger openStreamCount;
    BlockingQueue<Integer> buf = new ArrayBlockingQueue<Integer>(1);
    InputStream[] sources;

    public MergedInputStream(InputStream... sources) {
        this.sources = sources;
        openStreamCount = new AtomicInteger(sources.length);
        for (int i = 0; i < sources.length; i++)
            new ReadThread(i).start();
    }


    public void close() throws IOException {
        String ex = "";
        for (InputStream is : sources) {
            try {
                is.close();
            } catch (IOException e) {
                ex += e.getMessage() + " ";
            }
        }
        if (ex.length() > 0)
            throw new IOException(ex.substring(0, ex.length() - 1));
    }


    public int read() throws IOException {
        if (openStreamCount.get() == 0)
            return -1;

        try {
            return buf.take();
        } catch (InterruptedException e) {
            throw new IOException(e);
        }
    }


    private class ReadThread extends Thread {

        private final int src;
        public ReadThread(int src) {
            this.src = src;
        }

        public void run() {
            try {
                int data;
                while ((data = sources[src].read()) != -1)
                    buf.put(data);
            } catch (IOException ioex) {
            } catch (InterruptedException e) {
            }
            openStreamCount.decrementAndGet();
        }
    }
}

Answer 2

我可以想到三种方法来做到这一点：

• 使用非阻塞 I/O (API documentation)。这是最干净的解决方案。
• 多个线程，一个用于每个合并的输入流。线程将阻塞相关输入流的read() 方法，然后在数据可用时通知MergeInputStream 对象。 MergedInputStream 中的 read() 方法将等待此通知，然后从相应的流中读取数据。
• 具有繁忙循环的单线程。您的 MergeInputStream.read() 方法需要循环检查每个合并输入流的 available() 方法。如果没有可用数据，请休眠几毫秒。重复，直到数据在合并的输入流之一中可用。