Как преобразовать OutputStream в InputStream?

Я нахожусь на стадии разработки, где у меня есть два модуля, и из одного я получил вывод как OutputStreamи второй, который принимает только InputStream. Вы знаете , как конвертировать OutputStreamв InputStream( а не наоборот, я имею в виду на самом деле так) , что я буду иметь возможность соединить эти две части?

Спасибо

Это OutputStreamтот, где вы пишете данные. Если какой-то модуль предоставляет значение OutputStream, ожидается, что на другом конце что-то читается.

С InputStreamдругой стороны, что- то, что представляет собой , указывает на то, что вам нужно будет прослушать этот поток, и будут данные, которые вы сможете прочитать.

Таким образом, можно подключить InputStreamкOutputStream

InputStream----read---> intermediateBytes[n] ----write----> OutputStream

Как кто-то упомянул, это то, что позволяет copy()метод из IOUtils . Не имеет смысла идти другим путем ... надеюсь, это имеет какой-то смысл

ОБНОВИТЬ:

Конечно, чем больше я об этом думаю, тем больше я вижу, как это на самом деле будет требованием. Я знаю некоторые комментарии, упомянутые Pipedпотоки ввода / вывода, но есть и другая возможность.

Если выходной поток представляет собой ByteArrayOutputStream, то вы всегда можете получить полное содержимое, вызвав toByteArray()метод. Затем вы можете создать оболочку входного потока с помощью ByteArrayInputStreamподкласса. Эти два псевдопотока, оба они в основном просто заключают массив байтов. Таким образом, использование потоков таким образом технически возможно, но для меня это все еще очень странно ...

Кажется, что есть много ссылок и других подобных вещей, но нет реального кода, использующего каналы. Преимущество использования java.io.PipedInputStreamи java.io.PipedOutputStreamзаключается в том, что нет дополнительного потребления памяти. ByteArrayOutputStream.toByteArray()возвращает копию исходного буфера, так что это означает, что все, что у вас есть в памяти, теперь у вас есть две его копии. Затем запись в InputStreamсредство означает, что теперь у вас есть три копии данных.

Код:

// take the copy of the stream and re-write it to an InputStream
PipedInputStream in = new PipedInputStream();
final PipedOutputStream out = new PipedOutputStream(in);
new Thread(new Runnable() {
    public void run () {
        try {
            // write the original OutputStream to the PipedOutputStream
            // note that in order for the below method to work, you need
            // to ensure that the data has finished writing to the
            // ByteArrayOutputStream
            originalByteArrayOutputStream.writeTo(out);
        }
        catch (IOException e) {
            // logging and exception handling should go here
        }
        finally {
            // close the PipedOutputStream here because we're done writing data
            // once this thread has completed its run
            if (out != null) {
                // close the PipedOutputStream cleanly
                out.close();
            }
        }   
    }
}).start();

Этот код предполагает , что originalByteArrayOutputStreamэто , ByteArrayOutputStreamкак это, как правило , поток использоваться только выход, если вы пишете в файл. Надеюсь, это поможет! Самое замечательное в этом то, что, поскольку он находится в отдельном потоке, он также работает параллельно, поэтому все, что потребляет ваш входной поток, будет также вытекать из вашего старого выходного потока. Это полезно, потому что буфер может оставаться меньше, и у вас будет меньше задержка и меньше использования памяти.

Поскольку потоки ввода и вывода являются только начальной и конечной точкой, решение заключается во временном хранении данных в байтовом массиве. Таким образом, вы должны создать промежуточный продукт ByteArrayOutputStream, из которого вы создаете, byte[]который используется в качестве ввода для нового ByteArrayInputStream.

public void doTwoThingsWithStream(InputStream inStream, OutputStream outStream){ 
  //create temporary bayte array output stream
  ByteArrayOutputStream baos = new ByteArrayOutputStream();
  doFirstThing(inStream, baos);
  //create input stream from baos
  InputStream isFromFirstData = new ByteArrayInputStream(baos.toByteArray()); 
  doSecondThing(isFromFirstData, outStream);
}

Надеюсь, поможет.

Вам понадобится промежуточный класс, который будет буфер между. Каждый раз, когда InputStream.read(byte[]...)вызывается, класс буферизации будет заполнять переданный байтовый массив следующим портом, переданным из OutputStream.write(byte[]...). Поскольку размеры чанков могут не совпадать, класс адаптера должен будет хранить определенное количество до тех пор, пока ему не хватит для заполнения буфера чтения и / или возможности хранения любого переполнения буфера.

В этой статье есть несколько разных подходов к этой проблеме:

http://blog.ostermiller.org/convert-java-outputstream-inputstream

ByteArrayOutputStream buffer = (ByteArrayOutputStream) aOutputStream;
byte[] bytes = buffer.toByteArray();
InputStream inputStream = new ByteArrayInputStream(bytes);

Easystream библиотека с открытым исходным кодом имеет прямую поддержку для преобразования OutputStream к InputStream: http://io-tools.sourceforge.net/easystream/tutorial/tutorial.html

Они также перечисляют другие варианты: http://io-tools.sourceforge.net/easystream/OutputStream_to_InputStream.html

Я столкнулся с той же проблемой, связанной с преобразованием a ByteArrayOutputStreamв a, ByteArrayInputStreamи решил ее с помощью производного класса, из ByteArrayOutputStreamкоторого можно вернуть значение ByteArrayInputStream, инициализированное внутренним буфером объекта ByteArrayOutputStream. Таким образом, дополнительная память не используется, и «преобразование» происходит очень быстро:

package info.whitebyte.utils;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

/**
 * This class extends the ByteArrayOutputStream by 
 * providing a method that returns a new ByteArrayInputStream
 * which uses the internal byte array buffer. This buffer
 * is not copied, so no additional memory is used. After
 * creating the ByteArrayInputStream the instance of the
 * ByteArrayInOutStream can not be used anymore.
 * <p>
 * The ByteArrayInputStream can be retrieved using <code>getInputStream()</code>.
 * @author Nick Russler
 */
public class ByteArrayInOutStream extends ByteArrayOutputStream {
    /**
     * Creates a new ByteArrayInOutStream. The buffer capacity is
     * initially 32 bytes, though its size increases if necessary.
     */
    public ByteArrayInOutStream() {
        super();
    }

    /**
     * Creates a new ByteArrayInOutStream, with a buffer capacity of
     * the specified size, in bytes.
     *
     * @param   size   the initial size.
     * @exception  IllegalArgumentException if size is negative.
     */
    public ByteArrayInOutStream(int size) {
        super(size);
    }

    /**
     * Creates a new ByteArrayInputStream that uses the internal byte array buffer 
     * of this ByteArrayInOutStream instance as its buffer array. The initial value 
     * of pos is set to zero and the initial value of count is the number of bytes 
     * that can be read from the byte array. The buffer array is not copied. This 
     * instance of ByteArrayInOutStream can not be used anymore after calling this
     * method.
     * @return the ByteArrayInputStream instance
     */
    public ByteArrayInputStream getInputStream() {
        // create new ByteArrayInputStream that respects the current count
        ByteArrayInputStream in = new ByteArrayInputStream(this.buf, 0, this.count);

        // set the buffer of the ByteArrayOutputStream 
        // to null so it can't be altered anymore
        this.buf = null;

        return in;
    }
}

Я положил материал на GitHub: https://github.com/nickrussler/ByteArrayInOutStream

Библиотека io-extras может быть полезна. Например, если вы хотите сжать InputStreamиспользование, GZIPOutputStreamи вы хотите, чтобы это происходило синхронно (используя размер буфера по умолчанию 8192):

InputStream is = ...
InputStream gz = IOUtil.pipe(is, o -> new GZIPOutputStream(o));

Обратите внимание, что библиотека имеет 100% покрытие модульных тестов (что, конечно, стоит!) И находится в Maven Central. Зависимость Maven это:

<dependency>
  <groupId>com.github.davidmoten</groupId>
  <artifactId>io-extras</artifactId>
  <version>0.1</version>
</dependency>

Обязательно проверьте наличие более поздней версии.

С моей точки зрения, java.io.PipedInputStream / java.io.PipedOutputStream - лучший вариант для рассмотрения. В некоторых ситуациях вы можете использовать ByteArrayInputStream / ByteArrayOutputStream. Проблема в том, что вам нужно продублировать буфер для преобразования ByteArrayOutputStream в ByteArrayInputStream. Также ByteArrayOutpuStream / ByteArrayInputStream ограничены 2 ГБ. Вот реализация OutpuStream / InputStream, которую я написал, чтобы обойти ограничения ByteArrayOutputStream / ByteArrayInputStream (код Scala, но легко понятный для разработчиков Java):

import java.io.{IOException, InputStream, OutputStream}

import scala.annotation.tailrec

/** Acts as a replacement for ByteArrayOutputStream
  *
  */
class HugeMemoryOutputStream(capacity: Long) extends OutputStream {
  private val PAGE_SIZE: Int = 1024000
  private val ALLOC_STEP: Int = 1024

  /** Pages array
    *
    */
  private var streamBuffers: Array[Array[Byte]] = Array.empty[Array[Byte]]

  /** Allocated pages count
    *
    */
  private var pageCount: Int = 0

  /** Allocated bytes count
    *
    */
  private var allocatedBytes: Long = 0

  /** Current position in stream
    *
    */
  private var position: Long = 0

  /** Stream length
    *
    */
  private var length: Long = 0

  allocSpaceIfNeeded(capacity)

  /** Gets page count based on given length
    *
    * @param length   Buffer length
    * @return         Page count to hold the specified amount of data
    */
  private def getPageCount(length: Long) = {
    var pageCount = (length / PAGE_SIZE).toInt + 1

    if ((length % PAGE_SIZE) == 0) {
      pageCount -= 1
    }

    pageCount
  }

  /** Extends pages array
    *
    */
  private def extendPages(): Unit = {
    if (streamBuffers.isEmpty) {
      streamBuffers = new Array[Array[Byte]](ALLOC_STEP)
    }
    else {
      val newStreamBuffers = new Array[Array[Byte]](streamBuffers.length + ALLOC_STEP)
      Array.copy(streamBuffers, 0, newStreamBuffers, 0, streamBuffers.length)
      streamBuffers = newStreamBuffers
    }

    pageCount = streamBuffers.length
  }

  /** Ensures buffers are bug enough to hold specified amount of data
    *
    * @param value  Amount of data
    */
  private def allocSpaceIfNeeded(value: Long): Unit = {
    @tailrec
    def allocSpaceIfNeededIter(value: Long): Unit = {
      val currentPageCount = getPageCount(allocatedBytes)
      val neededPageCount = getPageCount(value)

      if (currentPageCount < neededPageCount) {
        if (currentPageCount == pageCount) extendPages()

        streamBuffers(currentPageCount) = new Array[Byte](PAGE_SIZE)
        allocatedBytes = (currentPageCount + 1).toLong * PAGE_SIZE

        allocSpaceIfNeededIter(value)
      }
    }

    if (value < 0) throw new Error("AllocSpaceIfNeeded < 0")
    if (value > 0) {
      allocSpaceIfNeededIter(value)

      length = Math.max(value, length)
      if (position > length) position = length
    }
  }

  /**
    * Writes the specified byte to this output stream. The general
    * contract for <code>write</code> is that one byte is written
    * to the output stream. The byte to be written is the eight
    * low-order bits of the argument <code>b</code>. The 24
    * high-order bits of <code>b</code> are ignored.
    * <p>
    * Subclasses of <code>OutputStream</code> must provide an
    * implementation for this method.
    *
    * @param      b the <code>byte</code>.
    */
  @throws[IOException]
  override def write(b: Int): Unit = {
    val buffer: Array[Byte] = new Array[Byte](1)

    buffer(0) = b.toByte

    write(buffer)
  }

  /**
    * Writes <code>len</code> bytes from the specified byte array
    * starting at offset <code>off</code> to this output stream.
    * The general contract for <code>write(b, off, len)</code> is that
    * some of the bytes in the array <code>b</code> are written to the
    * output stream in order; element <code>b[off]</code> is the first
    * byte written and <code>b[off+len-1]</code> is the last byte written
    * by this operation.
    * <p>
    * The <code>write</code> method of <code>OutputStream</code> calls
    * the write method of one argument on each of the bytes to be
    * written out. Subclasses are encouraged to override this method and
    * provide a more efficient implementation.
    * <p>
    * If <code>b</code> is <code>null</code>, a
    * <code>NullPointerException</code> is thrown.
    * <p>
    * If <code>off</code> is negative, or <code>len</code> is negative, or
    * <code>off+len</code> is greater than the length of the array
    * <code>b</code>, then an <tt>IndexOutOfBoundsException</tt> is thrown.
    *
    * @param      b   the data.
    * @param      off the start offset in the data.
    * @param      len the number of bytes to write.
    */
  @throws[IOException]
  override def write(b: Array[Byte], off: Int, len: Int): Unit = {
    @tailrec
    def writeIter(b: Array[Byte], off: Int, len: Int): Unit = {
      val currentPage: Int = (position / PAGE_SIZE).toInt
      val currentOffset: Int = (position % PAGE_SIZE).toInt

      if (len != 0) {
        val currentLength: Int = Math.min(PAGE_SIZE - currentOffset, len)
        Array.copy(b, off, streamBuffers(currentPage), currentOffset, currentLength)

        position += currentLength

        writeIter(b, off + currentLength, len - currentLength)
      }
    }

    allocSpaceIfNeeded(position + len)
    writeIter(b, off, len)
  }

  /** Gets an InputStream that points to HugeMemoryOutputStream buffer
    *
    * @return InputStream
    */
  def asInputStream(): InputStream = {
    new HugeMemoryInputStream(streamBuffers, length)
  }

  private class HugeMemoryInputStream(streamBuffers: Array[Array[Byte]], val length: Long) extends InputStream {
    /** Current position in stream
      *
      */
    private var position: Long = 0

    /**
      * Reads the next byte of data from the input stream. The value byte is
      * returned as an <code>int</code> in the range <code>0</code> to
      * <code>255</code>. If no byte is available because the end of the stream
      * has been reached, the value <code>-1</code> is returned. This method
      * blocks until input data is available, the end of the stream is detected,
      * or an exception is thrown.
      *
      * <p> A subclass must provide an implementation of this method.
      *
      * @return the next byte of data, or <code>-1</code> if the end of the
      *         stream is reached.
      */
    @throws[IOException]
    def read: Int = {
      val buffer: Array[Byte] = new Array[Byte](1)

      if (read(buffer) == 0) throw new Error("End of stream")
      else buffer(0)
    }

    /**
      * Reads up to <code>len</code> bytes of data from the input stream into
      * an array of bytes.  An attempt is made to read as many as
      * <code>len</code> bytes, but a smaller number may be read.
      * The number of bytes actually read is returned as an integer.
      *
      * <p> This method blocks until input data is available, end of file is
      * detected, or an exception is thrown.
      *
      * <p> If <code>len</code> is zero, then no bytes are read and
      * <code>0</code> is returned; otherwise, there is an attempt to read at
      * least one byte. If no byte is available because the stream is at end of
      * file, the value <code>-1</code> is returned; otherwise, at least one
      * byte is read and stored into <code>b</code>.
      *
      * <p> The first byte read is stored into element <code>b[off]</code>, the
      * next one into <code>b[off+1]</code>, and so on. The number of bytes read
      * is, at most, equal to <code>len</code>. Let <i>k</i> be the number of
      * bytes actually read; these bytes will be stored in elements
      * <code>b[off]</code> through <code>b[off+</code><i>k</i><code>-1]</code>,
      * leaving elements <code>b[off+</code><i>k</i><code>]</code> through
      * <code>b[off+len-1]</code> unaffected.
      *
      * <p> In every case, elements <code>b[0]</code> through
      * <code>b[off]</code> and elements <code>b[off+len]</code> through
      * <code>b[b.length-1]</code> are unaffected.
      *
      * <p> The <code>read(b,</code> <code>off,</code> <code>len)</code> method
      * for class <code>InputStream</code> simply calls the method
      * <code>read()</code> repeatedly. If the first such call results in an
      * <code>IOException</code>, that exception is returned from the call to
      * the <code>read(b,</code> <code>off,</code> <code>len)</code> method.  If
      * any subsequent call to <code>read()</code> results in a
      * <code>IOException</code>, the exception is caught and treated as if it
      * were end of file; the bytes read up to that point are stored into
      * <code>b</code> and the number of bytes read before the exception
      * occurred is returned. The default implementation of this method blocks
      * until the requested amount of input data <code>len</code> has been read,
      * end of file is detected, or an exception is thrown. Subclasses are encouraged
      * to provide a more efficient implementation of this method.
      *
      * @param      b   the buffer into which the data is read.
      * @param      off the start offset in array <code>b</code>
      *                 at which the data is written.
      * @param      len the maximum number of bytes to read.
      * @return the total number of bytes read into the buffer, or
      *         <code>-1</code> if there is no more data because the end of
      *         the stream has been reached.
      * @see java.io.InputStream#read()
      */
    @throws[IOException]
    override def read(b: Array[Byte], off: Int, len: Int): Int = {
      @tailrec
      def readIter(acc: Int, b: Array[Byte], off: Int, len: Int): Int = {
        val currentPage: Int = (position / PAGE_SIZE).toInt
        val currentOffset: Int = (position % PAGE_SIZE).toInt

        val count: Int = Math.min(len, length - position).toInt

        if (count == 0 || position >= length) acc
        else {
          val currentLength = Math.min(PAGE_SIZE - currentOffset, count)
          Array.copy(streamBuffers(currentPage), currentOffset, b, off, currentLength)

          position += currentLength

          readIter(acc + currentLength, b, off + currentLength, len - currentLength)
        }
      }

      readIter(0, b, off, len)
    }

    /**
      * Skips over and discards <code>n</code> bytes of data from this input
      * stream. The <code>skip</code> method may, for a variety of reasons, end
      * up skipping over some smaller number of bytes, possibly <code>0</code>.
      * This may result from any of a number of conditions; reaching end of file
      * before <code>n</code> bytes have been skipped is only one possibility.
      * The actual number of bytes skipped is returned. If <code>n</code> is
      * negative, the <code>skip</code> method for class <code>InputStream</code> always
      * returns 0, and no bytes are skipped. Subclasses may handle the negative
      * value differently.
      *
      * The <code>skip</code> method of this class creates a
      * byte array and then repeatedly reads into it until <code>n</code> bytes
      * have been read or the end of the stream has been reached. Subclasses are
      * encouraged to provide a more efficient implementation of this method.
      * For instance, the implementation may depend on the ability to seek.
      *
      * @param      n the number of bytes to be skipped.
      * @return the actual number of bytes skipped.
      */
    @throws[IOException]
    override def skip(n: Long): Long = {
      if (n < 0) 0
      else {
        position = Math.min(position + n, length)
        length - position
      }
    }
  }
}

Простота в использовании, нет дублирования буфера, нет 2 ГБ памяти

val out: HugeMemoryOutputStream = new HugeMemoryOutputStream(initialCapacity /*may be 0*/)

out.write(...)
...

val in1: InputStream = out.asInputStream()

in1.read(...)
...

val in2: InputStream = out.asInputStream()

in2.read(...)
...

Если вы хотите создать OutputStream из InputStream, есть одна основная проблема. Запись метода в блоки OutputStream до тех пор, пока это не будет сделано. Таким образом, результат доступен, когда метод записи закончен. Это имеет 2 последствия:

1. Если вы используете только один поток, вам нужно подождать, пока все не будет записано (поэтому вам нужно сохранить данные потока в памяти или на диске).
2. Если вы хотите получить доступ к данным до его завершения, вам нужен второй поток.

Вариант 1 может быть реализован с использованием байтовых массивов или в виде файла. Вариант 1 может быть реализован с использованием pipies (напрямую или с дополнительной абстракцией - например, RingBuffer или google lib из другого комментария).

Действительно, со стандартным Java нет другого способа решить проблему. Каждое решение является реализацией одного из них.

Существует одна концепция под названием «продолжение» ( подробности см. В Википедии ). В этом случае в основном это означает:

• есть специальный поток вывода, который ожидает определенное количество данных
• если достигнута сумма, поток передает управление своему аналогу, который является специальным входным потоком
• входной поток делает объем данных доступным до тех пор, пока он не будет прочитан, после чего он передает управление обратно в выходной поток

В то время как в некоторых языках эта концепция встроена, для Java нужна некоторая «магия». Например, "commons-javaflow" из apache реализует такое для java. Недостатком является то, что для этого требуются некоторые специальные модификации байт-кода во время сборки. Поэтому было бы разумно поместить все это в дополнительную библиотеку с помощью пользовательских скриптов сборки.

Старый пост, но может помочь другим, используйте этот способ:

OutputStream out = new ByteArrayOutputStream();
...
out.write();
...
ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(out.toString().getBytes()));

Хотя вы не можете преобразовать OutputStream в InputStream, java предоставляет способ с помощью PipedOutputStream и PipedInputStream, чтобы вы могли записывать данные в PipedOutputStream, чтобы они стали доступны через связанный PipedInputStream.
Некоторое время назад я столкнулся с подобной ситуацией, когда имел дело со сторонними библиотеками, которые требовали, чтобы им был передан экземпляр InputStream вместо экземпляра OutputStream.
Я решил эту проблему, используя PipedInputStream и PipedOutputStream.
Кстати, их сложно использовать, и вы должны использовать многопоточность, чтобы достичь того, что вы хотите. Недавно я опубликовал реализацию на github, которую вы можете использовать.
Вот ссылка . Вы можете пройти через вики, чтобы понять, как его использовать.