更好的PipedReader/PipedWriter替代方案？

Question

我需要一个缓冲的 char 流，其中一个线程写入数据，另一个线程从中读取。目前 我正在使用 PipedReader 和 PipedWriter，但这些类会导致性能问题：当 PipedReader 的内部缓冲区为空时，它会执行 wait(1000)，导致我的应用程序明显滞后。

是否有一些库与 PipedReader/PipedWriter 执行相同的功能，但具有更好的性能？还是我必须自己实现？

- Esko Luontola

---

这可能不是你想要的，但是你的写线程能在写完后通知读线程吗？并且当ready()为真时，读线程将继续读取，否则将休眠。 - Phil

请问您能否展示一下您的代码？wait(1000)不应该是问题，因为写入者在写入时会通知读取者。然后读取者就会从等待中跳出。 - tangens

谢谢，Phil。我现在不在家，所以无法测试它，但根据源代码，PipedWriter.flush()似乎会通知读者。我今天稍后会尝试一下。 - Esko Luontola

Answer 1

问题在于，当向PipedWriter中写入内容时，并不会自动通知PipedReader有数据可读。当尝试读取PipedReader的缓冲区并且为空时，PipedReader会循环等待，使用wait(1000)调用，直到缓冲区有数据可读。

解决方法是，在向管道中写入内容后始终调用PipedWriter.flush()。flush所做的只是在读取器上调用notifyAll()。修复有问题代码的方法看起来像这样。

（对我而言，PipedReader/PipedWriter的实现看起来非常像过早优化的情况——为什么不在每次写入时都进行notifyAll呢？另外读取器在一个主动循环中等待，每隔一秒钟醒来一次，而不是只有在有东西需要读取时才醒来。代码还包含一些todo注释，即它执行的读取器/写入器线程检测不够复杂。）

这个问题似乎也存在于PipedOutputStream中。在我的当前项目中，无法手动调用flush()（无法修改Commons IO的IOUtils.copy()），因此我通过创建低延迟包装器来修复它们。它们比原始类要好得多。 :-)

- Esko Luontola

Answer 2

将字符流 API 包装到 BlockingQueue 中应该相当容易。

不过，我必须说，PipedReader 使用轮询等待数据似乎相当反常。这个问题有文档记录吗？还是你自己发现的？

- Marcelo Cantos

---

我认为你想要链接到“http://java.sun.com/javase/6/docs/api/java/util/concurrent/BlockingQueue.html”。 - Phil

是的，那是我的备选计划，如果我找不到已经实现我所需功能的类。 - Esko Luontola

谢谢Phil。我用谷歌搜索太急了。:-) - Marcelo Cantos

确实有些反常。我发现了测试中的延迟，将其分析到PipedReader中，并发现原因是http://www.java2s.com/Open-Source/Java-Document/6.0-JDK-Core/io-nio/java/io/PipedReader.java.htm中硬编码的wait(1000)调用 - 一些旧的（自JDK1.1以来）Java标准库编码得很丑陋，这并不是第一次出现。 - Esko Luontola

仅为读者澄清：wait(1000) 不像 sleep(1000) - 当它被通知数据可用时，它会立即退出。超时是为了处理错误情况 - 特别是当数据源线程意外死亡并且无法再次通知时。 - Luke Usherwood

Answer 3

@Esko Luontola，我一直在阅读你在sbt包中的代码，试图理解你在做什么。看起来你想启动一个Process并向其中传递输入，并将操作结果分别传输到不同的位置。这是否正确？

我建议尝试修改ReaderToWriterCopier中的主循环，以便不再执行read()——这是一个阻塞操作，当涉及到PipedReader时会导致轮询——而是显式等待Writer执行flush。文档明确指出，flush会通知任何Reader。

我不确定如何运行你的代码，因此无法深入了解。希望这可以帮助你。

- Phil

---

我的使用案例是Process打印的内容有多个读者，每个读者都需要自己的流副本，从当前时刻开始，到读者关闭为止。不同的读者互不影响。例如，读取PipedReader的是OutputReader.waitForOutput()，写入者是ReaderToWriterCopier。主程序是SbtRunner，测试源码中使用它的是SbtRunnerTester。SBT代表http://code.google.com/p/simple-build-tool/。 - Esko Luontola

Answer 4

我实现了类似的东西，然后提出了一个问题，是否有其他人有更好、经过思考和测试的代码。

- Stephen Denne