PipedInputStream/PipedOutputStream 的缺陷

Question

我在 SO 上看到了两个答案，声称 Java 提供的 PipedInputStream 和 PipedOutputStream 类存在缺陷。但他们没有详细说明他们出了什么问题。他们真的有缺陷吗？如果有，以什么方式？我目前正在编写一些使用它们的代码，所以我想知道我是否走错路了。

One answer 说：

PipedInputStream 和 PipedOutputStream 已损坏（关于线程）。他们假设每个实例都绑定到一个特定的线程。这很奇怪。

在我看来，这既不奇怪也不破碎。或许作者心中还有一些其他的缺陷？

Another answer说：

在实践中，最好避免使用它们。我在 13 年中使用过一次，但我希望我没有。

但是那个作者记不起问题出在哪里了。

---

与所有类一样，尤其是在多线程中使用的类，如果您滥用它们，就会遇到问题。所以我不认为PipedInputStream 可能抛出的不可预测的"write end dead" IOException 是一个缺陷（未能连接到close() 是一个错误；有关更多信息，请参阅Daniel Ferbers 的文章Whats this? IOException: Write end dead） .还有哪些其他声称的缺陷？

Answer 1

它们没有缺陷。

与所有类一样，尤其是在多线程中使用的类，如果您滥用它们，就会遇到问题。 PipedInputStream 可以抛出的不可预测的“write end dead”IOException 不是缺陷（未能连接到 close() 是一个错误；请参阅 Daniel Ferbers 的文章 Whats this? IOException: Write end dead 以获取更多信息） .

Answer 2

我在我的项目中很好地使用了它们，它们对于动态修改流和传递它们非常有用。唯一的缺点似乎是 PipedInputStream 的缓冲区很短（大约 1024），而我的输出流在 8KB 左右。

它没有任何缺陷，而且效果很好。

-------- groovy 中的示例

public class Runner{


final PipedOutputStream source = new PipedOutputStream();
PipedInputStream sink = new PipedInputStream();

public static void main(String[] args) {
    new Runner().doit()
    println "Finished main thread"
}


public void doit() {

    sink.connect(source)

    (new Producer(source)).start()
    BufferedInputStream buffer = new BufferedInputStream(sink)
    (new Consumer(buffer)).start()
}
}

class Producer extends Thread {


OutputStream source
Producer(OutputStream source) {
    this.source=source
}

@Override
public void run() {

    byte[] data = new byte[1024];

    println "Running the Producer..."
    FileInputStream fout = new FileInputStream("/Users/ganesh/temp/www/README")

    int amount=0
    while((amount=fout.read(data))>0)
    {
        String s = new String(data, 0, amount);
        source.write(s.getBytes())
        synchronized (this) {
            wait(5);
        }
    }

    source.close()
}

}

class Consumer extends Thread{

InputStream ins

Consumer(InputStream ins)
{
    this.ins = ins
}

public void run()
{
    println "Consumer running"

    int amount;
    byte[] data = new byte[1024];
    while ((amount = ins.read(data)) >= 0) {
        String s = new String(data, 0, amount);
        println "< $s"
        synchronized (this) {
            wait(5);
        }
    }

}

}

Answer 3

一个缺陷可能是作者没有明确的方式向读者表明它遇到了问题：

PipedOutputStream out = new PipedOutputStream();
PipedInputStream in = new PipedInputStream(out);

new Thread(() -> {
    try {
        writeToOut(out);
        out.close();
    }
    catch (SomeDataProviderException e) {
        // Have to notify the reading side, but how?
    }
}).start();

readFromIn(in);

作者可以关闭out，但可能读者将其误解为数据结束。为了正确处理这个问题，需要额外的逻辑。如果提供手动断开管道的功能会更容易。

现在有JDK-8222924 请求手动断开管道的方法。

Answer 4

在我看来有一个缺陷。更准确地说，如果应该将数据泵入 PipedOutputStream 的线程在实际将单个字节写入流中之前过早死亡，则死锁的风险很高。这种情况下的问题是管道流的实现无法检测到损坏的管道。因此，从 PipedInputStream 读取的线程将在第一次调用 read() 时永远等待（即死锁）。

断管检测实际上依赖于对 write() 的第一次调用，因为实现会延迟初始化写入端线程，并且只有从那个时间点开始，断管检测才会起作用。

下面的代码重现了这种情况：

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;

import org.junit.Test;

public class PipeTest
{
    @Test
    public void test() throws IOException
    {
        final PipedOutputStream pout = new PipedOutputStream();
        PipedInputStream pin = new PipedInputStream();

        pout.connect(pin);

        Thread t = new Thread(new Runnable()
        {
            public void run()
            {
                try
                {
                    if(true)
                    {
                        throw new IOException("asd");
                    }
                    pout.write(0); // first byte which never get's written
                    pout.close();
                }
                catch(IOException e)
                {
                    throw new RuntimeException(e);
                }
            }
        });
        t.start();

        pin.read(); // wait's forever, e.g. deadlocks
    }
}

Answer 5

我在 JDK 实现中看到的缺陷是：

1) 没有超时，读写器可以无限阻塞。

2) 对何时传输数据的次优控制（应仅在刷新或循环缓冲区已满时进行）

所以我创建了自己的来解决上述问题，（通过 ThreadLocal 传递的超时值）：

PipedOutputStream

使用方法：

PiedOutputStreamTest

希望对你有帮助...