如何在 Java 中合并两个输入流？

Question

在 Java 中有两个 InputStream，有没有一种方法可以合并它们，从而以一个 InputStream 结束，从而为您提供两个流的输出？怎么样？

Answer 1

正如评论，不清楚你所说的合并是什么意思。

InputStream.available 并不一定会给您一个有用的答案和阻止流的行为，从而使“随机”从任何一个中获取可用输入变得复杂。您需要两个线程从流中读取数据，然后通过 java.io.Piped(In|Out)putStream 传回数据（尽管这些类有问题）。或者，对于某些类型的流，可以使用不同的接口，例如 java.nio 非阻塞通道。

如果您想要第一个输入流的完整内容，然后是第二个：new java.io.SequenceInputStream(s1, s2)。

Answer 2

java.io.SequenceInputStream 可能是您需要的。它接受流的枚举，并会输出第一个流的内容，然后是第二个，依此类推，直到所有流都为空。

Answer 3

您可以编写一个自定义的InputStream 实现来执行此操作。示例：

import java.io.IOException;
import java.io.InputStream;
import java.util.Collections;
import java.util.Deque;
import java.util.LinkedList;

public class CatInputStream extends InputStream {
    private final Deque<InputStream> streams;

    public CatInputStream(InputStream... streams) {
        this.streams = new LinkedList<InputStream>();
        Collections.addAll(this.streams, streams);
    }

    private void nextStream() throws IOException {
        streams.removeFirst().close();
    }

    @Override
    public int read() throws IOException {
        int result = -1;
        while (!streams.isEmpty()
                && (result = streams.getFirst().read()) == -1) {
            nextStream();
        }
        return result;
    }

    @Override
    public int read(byte b[], int off, int len) throws IOException {
        int result = -1;
        while (!streams.isEmpty()
                && (result = streams.getFirst().read(b, off, len)) == -1) {
            nextStream();
        }
        return result;
    }

    @Override
    public long skip(long n) throws IOException {
        long skipped = 0L;
        while (skipped < n && !streams.isEmpty()) {
            int thisSkip = streams.getFirst().skip(n - skipped);
            if (thisSkip > 0)
                skipped += thisSkip;
            else
                nextStream();
        }
        return skipped;
    }

    @Override
    public int available() throws IOException {
        return streams.isEmpty() ? 0 : streams.getFirst().available();
    }

    @Override
    public void close() throws IOException {
        while (!streams.isEmpty())
            nextStream();
    }
}

此代码未经测试，因此您的里程可能会有所不同。

Answer 4

我想不到。您可能必须将两个流的内容读入一个 byte[]，然后从中创建一个 ByteArrayInputStream。

Answer 5

这是一个特定于字节数组的 MVar 实现（确保添加您自己的包定义）。从这里开始，在合并流上编写输入流是微不足道的。如果需要，我也可以发布。

import java.nio.ByteBuffer;

public final class MVar {

  private static enum State {
    EMPTY, ONE, MANY
  }

  private final Object lock;

  private State state;

  private byte b;

  private ByteBuffer bytes;
  private int length;

  public MVar() {
    lock = new Object();
    state = State.EMPTY;
  }

  public final void put(byte b) {
    synchronized (lock) {
      while (state != State.EMPTY) {
        try {
          lock.wait();
        } catch (InterruptedException e) {}
      }
      this.b = b;
      state = State.ONE;
      lock.notifyAll();
    }
  }

  public final void put(byte[] bytes, int offset, int length) {
    if (length == 0) {
      return;
    }
    synchronized (lock) {
      while (state != State.EMPTY) {
        try {
          lock.wait();
        } catch (InterruptedException e) {}
      }
      this.bytes = ByteBuffer.allocateDirect(length);
      this.bytes.put(bytes, offset, length);
      this.bytes.position(0);
      this.length = length;
      state = State.MANY;
      lock.notifyAll();
    }
  }

  public final byte take() {
    synchronized (lock) {
      while (state == State.EMPTY) {
        try {
          lock.wait();
        } catch (InterruptedException e) {}
      }
      switch (state) {
      case ONE: {
        state = State.EMPTY;
        byte b = this.b;
        lock.notifyAll();
        return b;
      }
      case MANY: {
        byte b = bytes.get();
        state = --length <= 0 ? State.EMPTY : State.MANY;
        lock.notifyAll();
        return b;
      }
      default:
        throw new AssertionError();
      }
    }
  }

  public final int take(byte[] bytes, int offset, int length) {
    if (length == 0) {
      return 0;
    }
    synchronized (lock) {
      while (state == State.EMPTY) {
        try {
          lock.wait();
        } catch (InterruptedException e) {}
      }
      switch (state) {
      case ONE:
        bytes[offset] = b;
        state = State.EMPTY;
        lock.notifyAll();
        return 1;
      case MANY:
        if (this.length > length) {
          this.bytes.get(bytes, offset, length);
          this.length = this.length - length;
          synchronized (lock) {
            lock.notifyAll();
          }
          return length;
        }
        this.bytes.get(bytes, offset, this.length);
        this.bytes = null;
        state = State.EMPTY;
        length = this.length;
        lock.notifyAll();
        return length;
      default:
        throw new AssertionError();
      }
    }
  }
}