Как объединить два потока ввода в Java?

Ответ 1

Как прокомментировано, неясно, что вы подразумеваете под слиянием.

Принимая доступный вход "случайно" из либо сложный с помощью InputStream.available, не обязательно дает вам полезный ответ и блокирует поведение потоков. Вам потребуется два потока для чтения из потоков, а затем передача данных через, скажем, java.io.Piped(In|Out)putStream (хотя эти классы имеют проблемы). В качестве альтернативы для некоторых типов потоков может быть возможно использовать другой интерфейс, например java.nio неблокирующие каналы.

Если вы хотите, чтобы полное содержимое первого входного потока сопровождалось вторым: new java.io.SequenceInputStream(s1, s2).

Ответ 2

java.io.SequenceInputStream может быть то, что вам нужно. Он принимает перечисление потоков и выводит содержимое первого потока, затем второго и так далее, пока все потоки не будут пустыми.

Ответ 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();
    }
}

Этот код не проверен, поэтому ваш пробег может отличаться.

Ответ 4

Не то, чтобы я мог думать. Вам, вероятно, придется прочитать содержимое двух потоков в байт [], а затем создать ByteArrayInputStream из этого.

Ответ 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();
      }
    }
  }
}