java - Javaで2つの入力ストリームをどのようにマージしますか?

Question

Java に 2 つの InputStream がある場合、それらをマージして、両方のストリームの出力を提供する 1 つの InputStream で終了する方法はありますか? どのように？

Answer 1

コメントしたように、マージの意味が明確ではありません。

いずれかから利用可能な入力を「ランダムに」取得InputStream.availableすることは、必ずしも有用な回答を提供するとは限らず、ストリームの動作をブロックするため複雑です。たとえば、ストリームから読み取り、データを戻すには 2 つのスレッドが必要ですjava.io.Piped(In|Out)putStream(ただし、これらのクラスには問題があります)。java.nioあるいは、ストリームの種類によっては、ノンブロッキング チャネルなど、別のインターフェイスを使用できる場合もあります。

最初の入力ストリームの完全なコンテンツとそれに続く 2 番目の入力ストリームが必要な場合: new java.io.SequenceInputStream(s1, s2).

Answer 2

java.io.SequenceInputStreamあなたが必要なものかもしれません。ストリームの列挙を受け入れ、最初のストリームの内容を出力し、次に2番目のストリームを出力し、すべてのストリームが空になるまで続けます。

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

これは、バイト配列に固有の 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();
      }
    }
  }
}

Answer 5

私が考えることができるわけではありません。おそらく、2 つのストリームの内容を byte[] に読み込み、そこから ByteArrayInputStream を作成する必要があります。