在写入时从 ByteArrayOutputStream 中读取

Question

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

网上看到的例子说用PipedOutputStream。如果我的第一个线程使 ByteArrayOutputStream 从一个单独的可重用库中对外界可用，我不知道如何将 inputStream 连接到它。

如何设置 PipedInputStream 以将其连接到 ByteArrayOutputStream 以从中读取数据？此外，当从 ByteArrayOutputStream 读取最后一个块时，我是否会看到 bytesRead == -1，指示 outputStream 何时从第一个线程关闭？

非常感谢，迈克

Answer 1

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

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

当 PipedInputStream 中的缓冲区填满时写入 PipedOutputStream 将阻塞，当缓冲区为空时从 PipedInputStream 读取将阻塞，因此生产者线程将如果它得到 "ahead" 消费者的睡眠（块），反之亦然。

阻塞线程后等待 1000 毫秒，然后再检查缓冲区，因此最好在写入完成后刷新输出（如果 reader 正在休眠，这将唤醒它）。

当您在生产者线程中关闭输出流时，您的输入流将看到 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();
    }
}

在写入时从 ByteArrayOutputStream 中读取

Read from ByteArrayOutputStream while it's being written to

java

multithreading

inputstream

bytearrayoutputstream