在写入时从 ByteArrayOutputStream 读取

Posted 2023-02-25

【中文标题】在写入时从 ByteArrayOutputStream 读取

【英文标题】：Read from ByteArrayOutputStream while it's being written to

【发布时间】：2018-03-20 04:02:04

【问题描述】：

我有一个类不断生成数据并将其写入自己线程上的 ByteArrayOutputStream。我有一个第二个线程获取对此 ByteArrayOutputStream 的引用。我希望第二个线程读取 ByteArrayOutputStream 的任何数据（并清空），然后在它没有得到任何字节并休眠时停止。睡眠后，我希望它尝试获取更多数据并再次清空。

我在网上看到的例子都说使用 PipedOutputStream。如果我的第一个线程是让 ByteArrayOutputStream 从一个单独的可重用库提供给外部世界，我不知道如何将 inputStream 连接到它。

如何设置 PipedInputStream 以将其连接到 ByteArrayOutputStream 以从上面读取？另外，当从 ByteArrayOutputStream 读取最后一个块时，我会看到 bytesRead == -1，表示 outputStream 何时从第一个线程关闭？

非常感谢， 迈克

【参考方案1】：

直接写入PipedOutputStream（也就是说，根本不要使用ByteArrayOutputStream）。它们都扩展了OutputStream，因此具有相同的接口。

PipedOutputStream 和 PipedInputStream 中都有 connect 方法，用于将两个管道连接在一起，或者您可以使用其中一个构造函数来创建一对。

当PipedInputStream 中的缓冲区填满时，写入PipedOutputStream 将阻塞，而当缓冲区为空时，从PipedInputStream 读取将阻塞，因此生产者线程如果得到“将休眠（阻塞）”领先于消费者”，反之亦然。

在阻塞线程后等待 1000 毫秒再重新检查缓冲区，因此最好在写入完成后刷新输出（如果阅读器处于睡眠状态，这将唤醒阅读器）。

当您在生产者线程中关闭输出流时，您的输入流将看到 EOF (bytesRead == -1)。

import java.io.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class PipeTest 
    public static void main(String[] args) throws IOException 
        PipedOutputStream out = new PipedOutputStream();
        // Wire an input stream to the output stream, and use a buffer of 2048 bytes
        PipedInputStream in = new PipedInputStream(out, 2048);

        ExecutorService executor = Executors.newCachedThreadPool();

        // Producer thread.
        executor.execute(() -> 
            try 
                for (int i = 0; i < 10240; i++) 
                    out.write(0);
                    // flush to wake the reader
                    out.flush();
                
                out.close();
             catch (IOException e) 
                throw new UncheckedIOException(e);
            
        );

        // Consumer thread.
        executor.execute(() -> 
            try 
                int b, read = 0;
                while ((b = in.read()) != -1) 
                    read++;
                
                System.out.println("Read " + read + " bytes.");
             catch (IOException e) 
                throw new UncheckedIOException(e);
            
        );

        executor.shutdown();
    

【讨论】：

太棒了。感谢您为我解决这个问题！

嗨，teppic，我已经编写了所有这些代码，似乎我的生产者线程必须在读取开始之前填充连接的 PipedInputStream 的缓冲区。这是有道理的，但这是否意味着如果我的生产者正在从麦克风输出数据，是否会有一段时间丢失输入？我应该考虑某种“双缓冲”来处理这个问题吗？

尝试在每次写入后刷新输出流。