OutputStream
,第二个模块仅接受InputStream
。您是否知道如何将OutputStream
转换为InputStream
(反之亦然,实际上是这样),我才能将这两部分连接起来?谢谢
#1 楼
OutputStream
是您向其中写入数据的地方。如果某个模块公开了OutputStream
,则期望在另一端读取内容。另一方面,公开了
InputStream
的内容表明您需要侦听此流,并且会有一些您可以读取的数据。 因此可以将
InputStream
连接到OutputStream
InputStream----read---> intermediateBytes[n] ----write----> OutputStream
正如有人提到的那样,这就是IOUtils的
copy()
方法可以让您做。换种方式没有道理...希望这有道理UPDATE:
当然,我越想这个,我就能看到的越多这实际上是如何要求的。我知道提到了
Piped
输入/输出流的一些注释,但是还有另一种可能性。如果公开的输出流是
ByteArrayOutputStream
,则始终可以通过调用toByteArray()
来获取全部内容方法。然后,可以使用ByteArrayInputStream
子类创建输入流包装器。这两个都是伪流,它们基本上都只是包装一个字节数组。因此,以这种方式使用流在技术上是可行的,但对我来说仍然很奇怪... 评论
copy()根据API将其执行到OS,我需要它向后执行
– Waypoint
2011-4-25 13:47
用例非常简单:假设您有一个序列化库(例如,序列化为JSON)和一个使用InputStream的传输层(例如,Tomcat)。因此,您需要通过HTTP连接从JSON传递来自JSON的OutputStream,该HTTP连接要从InputStream读取。
– JBCP
2014年2月21日在17:28
当进行单元测试时,这很有用,并且您对避免接触文件系统非常着迷。
–乔恩
2014年3月13日在9:26
@JBCP的评论是当场。另一个用例是在HTTP请求期间使用PDFBox生成PDF。 PDFBox使用OutputStream保存PDF对象,并且REST API接受InputStream来回复客户端。因此,OutputStream-> InputStream是一个非常实际的用例。
–约翰·曼科(John Manko)
15年9月12日在18:47
“您始终可以通过调用toByteArray()方法获取完整的内容”,使用流的要点是不要将整个内容加载到内存中!
– stackoverflowd
20年4月8日在7:06
#2 楼
似乎有许多链接和其他类似内容,但没有使用管道的实际代码。使用java.io.PipedInputStream
和java.io.PipedOutputStream
的优点是没有额外的内存消耗。 ByteArrayOutputStream.toByteArray()
返回原始缓冲区的副本,这意味着内存中的任何内容现在都有两个副本。然后写入InputStream
意味着您现在具有三个数据副本。代码:
originalByteArrayOutputStream
,因为它通常是唯一可用的输出流,除非您要写入文件。我希望这有帮助!这样做的好处在于,由于它在单独的线程中,因此也可以并行工作,因此,消耗您输入流的任何内容也将从您的旧输出流中流出。这是有好处的,因为缓冲区可以保持较小,并且您将拥有更少的延迟和更少的内存使用。评论
我对此表示赞成,但最好传递给in的构造函数,否则您可能会由于竞态条件(我经历过)而收到封闭管道异常。使用Java 8 Lambdas:PipedInputStream in = new PipedInputStream(out); (((Runnable)()-> {originalOutputStream.writeTo(out);})。run();返回
–约翰·曼科(John Manko)
2015年9月13日下午5:07
@JohnManko嗯...我从来没有遇到过这个问题。您是否因为其他线程或主线程正在调用out.close()而遇到了这种情况?的确,此代码假定您的PipedOutputStream的寿命比应为true的原始OutputStream的寿命长,但没有假定您如何控制流。这留给开发人员。这段代码中没有任何东西会导致封闭或损坏的管道异常。
–mikeho
2015年9月14日在17:35
不,我的情况源于我将PDF存储在Mongo GridFS中,然后使用Jax-RS流传输到客户端的情况。 MongoDB提供了OutputStream,但是Jax-RS需要InputStream。我的path方法似乎会在OutputStream完全建立之前返回带有InputStream的容器(也许缓冲区尚未被缓存)。无论如何,Jax-RS会在InputStream上引发管道关闭异常。奇怪,但这就是一半时间发生的情况。更改为上面的代码可以防止这种情况。
–约翰·曼科(John Manko)
2015年9月14日在21:12
@JohnManko我正在研究更多内容,并且从PipedInputStream Javadocs中看到:如果为连接的管道输出流提供数据字节的线程不再存在,则据说管道已损坏。因此,我怀疑如果您使用的是上面的示例,则该线程在Jax-RS使用输入流之前就已经完成了。同时,我查看了MongoDB Javadocs。 GridFSDBFile具有输入流,那么为什么不将其传递给Jax-RS?
–mikeho
2015年9月15日在16:44
@DennisCheung是的,当然。没有什么是免费的,但是肯定会小于15MB。优化将包括使用线程池来减少具有恒定线程/对象创建的GC流失率。
–mikeho
16年7月27日在18:53
#3 楼
由于输入和输出流只是起点和终点,因此解决方案是将数据临时存储在字节数组中。因此,您必须创建中间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);
}
希望对您有所帮助。
评论
baos.toByteArray()使用System.arraycopy创建一个副本。感谢@mikeho指出了developer.classpath.org/doc/java/io/…
– Mitja Gustin
18/12/5在20:30
#4 楼
ByteArrayOutputStream buffer = (ByteArrayOutputStream) aOutputStream;
byte[] bytes = buffer.toByteArray();
InputStream inputStream = new ByteArrayInputStream(bytes);
评论
您不应该使用此方法,因为toByteArray()方法主体类似于此返回Arrays.copyOf(buf,count);。返回一个新数组。
–根G
18年5月19日在11:29
无法将java.io.FileOutputStream强制转换为java.io.ByteArrayOutputStream
– kAmol
20 Dec 16'在6:10
#5 楼
您将需要一个中间类,将在它们之间进行缓冲。每次调用InputStream.read(byte[]...)
时,缓冲类都会使用从OutputStream.write(byte[]...)
传入的下一个块填充传入的字节数组。由于块的大小可能不相同,因此适配器类将需要存储一定数量,直到它有足够的空间来填充读取缓冲区和/或能够存储任何缓冲区溢出为止。本文详细介绍了解决此问题的几种不同方法:
http://blog.ostermiller.org/convert-java-outputstream-inputstream
评论
谢谢@mckamey,基于循环缓冲区的方法正是我所需要的!
–王辉
16年4月7日在14:25
#6 楼
easystream开源库直接支持将OutputStream转换为InputStream:http://io-tools.sourceforge.net/easystream/tutorial/tutorial.html // create conversion
final OutputStreamToInputStream<Void> out = new OutputStreamToInputStream<Void>() {
@Override
protected Void doRead(final InputStream in) throws Exception {
LibraryClass2.processDataFromInputStream(in);
return null;
}
};
try {
LibraryClass1.writeDataToTheOutputStream(out);
} finally {
// don't miss the close (or a thread would not terminate correctly).
out.close();
}
它们还列出了其他选项:http://io-tools.sourceforge.net/easystream/outputstream_to_inputstream/implementations.html
将数据写入数据内存缓冲区(ByteArrayOutputStream)获取byteArray并使用ByteArrayInputStream再次读取它。如果您确定数据适合内存,这是最好的方法。
将数据复制到临时文件中并读回。
使用管道:这是内存使用和速度方面的最佳方法(您可以充分利用多核处理器)以及Sun提供的标准解决方案。
使用easystream库中的InputStreamFromOutputStream和OutputStreamToInputStream。
评论
是的,使用easystream!
–smartwjw
2015年9月23日下午4:49
#7 楼
我在将ByteArrayOutputStream
转换为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
评论
如果内容不适合缓冲区怎么办?
–Vadimo
2014-09-15 9:20
然后,您不应该首先使用ByteArrayInputStream。
–尼克·罗斯勒(Nick Russler)
2014-09-15 9:39
此解决方案将在内存中存储所有字节。对于小文件,这可以,但是您也可以在ByteArrayOutput流上使用getBytes()
–Vadimo
2014年9月15日上午11:40
如果您的意思是toByteArray,这将导致内部缓冲区被复制,这将占用我的方法两倍的内存。编辑:嗯,我知道,对于小文件,这当然可以工作。
–尼克·罗斯勒(Nick Russler)
2014年9月15日15:40
浪费时间。 ByteArrayOutputStream具有writeTo方法,可将内容传输到另一个输出流
– Tony BenBrahim
15年2月17日在18:38
#8 楼
库io-extras可能有用。例如,如果要使用InputStream
对GZIPOutputStream
进行gzip压缩,并且希望它同步发生(使用默认缓冲区大小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>
请确保检查更高版本。
#9 楼
从我的角度来看,java.io.PipedInputStream / java.io.PipedOutputStream是考虑的最佳选择。在某些情况下,您可能需要使用ByteArrayInputStream / ByteArrayOutputStream。问题是您需要复制缓冲区以将ByteArrayOutputStream转换为ByteArrayInputStream。 ByteArrayOutpuStream / ByteArrayInputStream也限制为2GB。这是我编写的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
}
}
}
}
易于使用,没有缓冲区复制,没有2GB内存限制
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(...)
...
#10 楼
如果要从InputStream生成OutputStream,则存在一个基本问题。写入OutputStream的方法将阻塞直到完成。因此,写入方法完成后即可得到结果。这有2个结果:如果仅使用一个线程,则需要等到所有内容都写完之后(因此需要将流的数据存储在内存或磁盘中)。 >如果要在完成数据之前访问数据,则需要第二个线程。
变量1可以使用字节数组或字段来实现。
变量1可以使用pipies(直接或额外的抽象-例如,RingBuffer或其他评论中的google lib)。
与标准Java无关,没有其他方法可以解决此问题。每个解决方案都是其中一种的实现。
有一个概念称为“继续”(有关详细信息,请参阅Wikipedia)。在这种情况下,这基本上意味着:
有一个特殊的输出流,它期望一定数量的数据
,如果达到该数量,该流将控制它的对应对象,是一种特殊的输入流
输入流使可用的数据量直到被读取,之后,它将控制权传递回输出流
虽然某些语言已构建了此概念在,对于Java,您需要一些“魔术”。例如,来自apache的“ commons-javaflow”实现了Java。缺点是,这需要在构建时进行一些特殊的字节码修改。因此,将所有内容放入带有自定义构建脚本的额外库中是很有意义的。
#11 楼
旧帖子,但可能会帮助其他人,请使用以下方式:OutputStream out = new ByteArrayOutputStream();
...
out.write();
...
ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(out.toString().getBytes()));
评论
到字符串->大小问题
–user1594895
2014年10月9日下午13:11
此外,在流*上调用toString()。getBytes()不会返回流的内容。
–马滕·博德威斯(Maarten Bodewes)
2016年9月6日23:03
评论
参见stackoverflow.com/questions/1225909/…@ c0mrade,操作员仅在另一个方向上需要IOUtils.copy之类的东西。当某人写入OutputStream时,其他人可以在InputStream中使用它。基本上,这就是PipedOutputStream / PipedInputStream所做的。不幸的是,不能从其他流中构建管道流。
那么PipedOutputStream / PipedInputStream是解决方案吗?
基本上,为了使PipedStreams能够在您的情况下工作,您的OutputStream必须像新的YourOutputStream(thePipedOutputStream)和new YourInputStream(thePipedInputStream)那样构造,这可能不是您的流的工作方式。所以我不认为这是解决方案。