从输出流中获取输入流

Question

我有一个组件在输出流 (ByteArrayOutputStream) 中为我提供数据，我需要将其写入 SQL 数据库的 blob 字段而不创建临时缓冲区，因此需要获取输入流。

根据here 和here 的回答，我想出了以下方法从输出流中获取输入流：

private PipedInputStream getInputStream(ByteArrayOutputStream outputStream) throws InterruptedException
{
    PipedInputStream pipedInStream = new PipedInputStream();
    Thread copyThread = new Thread(new CopyStreamHelper(outputStream, pipedInStream));
    copyThread.start();
    // Wait for copy to complete
    copyThread.join();
    return pipedInStream;
}

class CopyStreamHelper implements Runnable
{
    private ByteArrayOutputStream outStream;
    private PipedInputStream pipedInStream;

    public CopyStreamHelper (ByteArrayOutputStream _outStream, PipedInputStream _pipedInStream)
    {
        outStream = _outStream;
        pipedInStream = _pipedInStream;
    }

    public void run()
    {
        PipedOutputStream pipedOutStream = null;
        try
        {
            // write the original OutputStream to the PipedOutputStream
            pipedOutStream = new PipedOutputStream(pipedInStream);
            outStream.writeTo(pipedOutStream);
        }
        catch (IOException e) 
        {
            // logging and exception handling should go here
        }
        finally
        {
            IOUtils.closeQuietly(pipedOutStream);
        }
    }
}

请注意，输出流已经包含写入的数据，它最多可以运行 1-2 MB。 然而，无论尝试在两个单独的线程或同一个线程中执行此操作，我发现总是PipedInputStream 挂在以下位置：

Object.wait(long) line: not available [native method]   
PipedInputStream.awaitSpace() line: not available   

Answer 1

您的解决方案过于复杂

ByteArrayOutputStream baos = ...;
byte[] data = baos.toByteArray();
return new ByteArrayInputStream(data);

Answer 2

在某种程度上必须有一个缓冲区。见Connecting an input stream to an outputstream。

我最喜欢的答案来自 Dean Hiller：

void feedInputToOutput(InputStream in, OutputStream out) {
   IOUtils.copy(in, out);
}

详情请见the api

Answer 3

我制作了一个非常简单的使用 PipedInput/OutputStream 的演示。它可能适合也可能不适合您的用例。

写入 PipedOutputStream 的生产类：

public class Producer implements Runnable {

    private final PipedOutputStream pipedOutputStream;
    private final PipedInputStream pipedInputStream;

    public Producer() throws IOException {
        this.pipedOutputStream = new PipedOutputStream();
        this.pipedInputStream = new PipedInputStream(pipedOutputStream);
    }

    public PipedInputStream getPipedInputStream() {
        return pipedInputStream;
    }

    @Override
    public void run() {

        try(InputStream inputStream = ByteStreams.limit(new RandomInputStream(), 100000)) {
            // guava copy function
            ByteStreams.copy(inputStream, pipedOutputStream);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                pipedOutputStream.close();
            } catch (IOException e) {
                // no-op
            }
        }
    }

    public static void main(String[] args) throws IOException {

        try {
            Producer producer = new Producer();
            Consumer consumer = new Consumer(producer);

            Thread thread1 = new Thread(producer);
            Thread thread2 = new Thread(consumer);

            thread1.start();
            thread2.start();

            thread1.join();
            thread2.join();

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

只计算字节数的消费者：

public class Consumer implements Runnable {

    private final Producer producer;

    public Consumer(Producer producer) {
        this.producer = producer;
    }

    @Override
    public void run() {

        try (PipedInputStream pipedInputStream = producer.getPipedInputStream()) {

            int counter = 0;
            while (pipedInputStream.read() != -1) {
                counter++;
            }

            System.out.println(counter);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}