代码之家 › 专栏 › 技术社区 › Alex Abdugafarov

Java:合并输入流

inputstream merge multithreading java

Alex Abdugafarov · 技术社区 · 15 年前

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

2 回复 | 直到 12 年前

Grodriguez 15 年前

我可以想出三种方法:

使用非阻塞I/O( API documentation ). 这是最干净的解决方案。
多个线程,每个合并的输入流一个线程。线程会阻塞 read() MergeInputStream 当数据可用时。这个 读取() 方法 MergedInputStream 将等待此通知,然后从相应流中读取数据。
有繁忙循环的单线程。你的 MergeInputStream.read() available() 每个合并输入流的方法。如果没有可用的数据,则休眠几毫秒。重复此操作,直到合并的输入流之一中的数据可用为止。

aioobe 15 年前

最好解决从多个源读取输入并将其序列化为一个流的问题 使用 SelectableChannel 和 Selector

如果没有可选择的频道,您可以选择 用一根线解决它 通过让read实现执行以下操作:为每个输入流 is ,检查是否 is.available() > 0 ,如果是,请返回 is.read() . 重复此过程,直到某个输入流有可用的数据。

Not all implementations of InputStream 工具 available() 在某种程度上,它返回0 if和only if read() 会阻塞的。结果是,很自然地,数据可能不会从这个流中读取,即使 正在读取() 会返回一个值。这是否被认为是一个bug是值得怀疑的,因为文档仅仅声明它应该返回可用字节数的“估计值”。
它使用所谓的“忙循环”,这基本上意味着你要么需要在循环中加入睡眠(这会导致读取延迟),要么不必要地占用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();
        }
    }
}