在写入时从 ByteArrayOutputStream 读取

Question

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

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

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

非常感谢， 迈克

Answer 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();
    }
}