对于在 Java 中创建一个 既是 InputStream 又是 OutputStream 的 Pipe 对象，是否有人有任何好的建议，因为 Java 没有多重继承，并且两个流都是抽象类而不是接口？
基本需求是拥有一个可以传递给需要 InputStream 或 OutputStream 的事物的单一对象，以将输出从一个线程传递到另一个线程的输入。
【问题讨论】：
标签： java input multiple-inheritance
最好使用 Pipe 或 ArrayBlockingQueue，我建议您不要使用 PipedInput/OutputStream，因为它们的做法很糟糕，即使您可以在下面的链接中看到它们已要求弃用，因为这会导致很多问题。
https://bugs.openjdk.java.net/browse/JDK-8223048
对于 BlockingQueue 和 Pipe，这里是一个简单的例子
管道：
Pipe pipe = Pipe.open();
Pipe.SinkChannel sinkChannel = pipe.sink();
String newData = "New String to write to file..." + System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while(buf.hasRemaining()) {
sinkChannel.write(buf);
}
Pipe.SourceChannel sourceChannel = pipe.source();
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = inChannel.read(buf);
参考：http://tutorials.jenkov.com/java-nio/pipe.html
阻塞队列：
//Shared class used by threads
public class Buffer {
// ArrayBlockingQueue
private BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(1);
public void get() {
// retrieve from ArrayBlockingQueue
try {
System.out.println("Consumer received - " + blockingQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void put(int data) {
try {
// putting in ArrayBlockingQueue
blockingQueue.put(data);
System.out.println("Producer produced - " + data);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
// Starting two threads
ExecutorService executorService = null;
try {
Buffer buffer = new Buffer();
executorService = Executors.newFixedThreadPool(2);
executorService.execute(new Producer(buffer));
executorService.execute(new Consumer(buffer));
} catch (Exception e) {
e.printStackTrace();
}finally {
if(executorService != null) {
executorService.shutdown();
}
}
}
public class Consumer implements Runnable {
private Buffer buffer;
public Consumer(Buffer buffer) {
this.buffer = buffer;
}
@Override
public void run() {
while (true) {
try {
buffer.get();
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public class Producer implements Runnable {
private Buffer buffer;
public Producer(Buffer buffer) {
this.buffer = buffer;
}
@Override
public void run() {
while (true) {
Random random = new Random();
int data = random.nextInt(1000);
buffer.put(data);
}
}
}
参考： https://github.com/kishanjavatrainer/ArrayBlockingQueueDemo/tree/master/ArrayBlockingQueueDemo
【讨论】：
我必须为与 Servlet 的慢速连接实现一个过滤器，所以基本上我将 servlet 输出流包装到一个 QueueOutputStream 中，它将每个字节（在小缓冲区中）添加到一个队列中，然后将这些小缓冲区输出到第二个输出流，因此在某种程度上它充当输入/输出流，恕我直言，这比无法很好扩展的 JDK 管道更好，基本上标准 JDK 实现（每个读/写）中有太多的上下文切换，阻塞队列非常适合单个生产者/消费者场景：
import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.*;
public class QueueOutputStream extends OutputStream
{
private static final int DEFAULT_BUFFER_SIZE=1024;
private static final byte[] END_SIGNAL=new byte[]{};
private final BlockingQueue<byte[]> queue=new LinkedBlockingDeque<>();
private final byte[] buffer;
private boolean closed=false;
private int count=0;
public QueueOutputStream()
{
this(DEFAULT_BUFFER_SIZE);
}
public QueueOutputStream(final int bufferSize)
{
if(bufferSize<=0){
throw new IllegalArgumentException("Buffer size <= 0");
}
this.buffer=new byte[bufferSize];
}
private synchronized void flushBuffer()
{
if(count>0){
final byte[] copy=new byte[count];
System.arraycopy(buffer,0,copy,0,count);
queue.offer(copy);
count=0;
}
}
@Override
public synchronized void write(final int b) throws IOException
{
if(closed){
throw new IllegalStateException("Stream is closed");
}
if(count>=buffer.length){
flushBuffer();
}
buffer[count++]=(byte)b;
}
@Override
public synchronized void write(final byte[] b, final int off, final int len) throws IOException
{
super.write(b,off,len);
}
@Override
public synchronized void close() throws IOException
{
flushBuffer();
queue.offer(END_SIGNAL);
closed=true;
}
public Future<Void> asyncSendToOutputStream(final ExecutorService executor, final OutputStream outputStream)
{
return executor.submit(
new Callable<Void>()
{
@Override
public Void call() throws Exception
{
try{
byte[] buffer=queue.take();
while(buffer!=END_SIGNAL){
outputStream.write(buffer);
buffer=queue.take();
}
outputStream.flush();
} catch(Exception e){
close();
throw e;
} finally{
outputStream.close();
}
return null;
}
}
);
}
【讨论】：
似乎忽略了这个问题的重点。如果我对您的理解正确，您需要一个对象在一个线程中像 InputStream 一样工作，而在另一个线程中像 OutputStream 一样创建一种在两个线程之间进行通信的方法。
也许一个答案是使用组合而不是继承（无论如何这是推荐的做法）。创建一个包含相互连接的 PipedInputStream 和 PipedOutputStream 的管道，使用 getInputStream() 和 getOutputStream() 方法。
你不能直接将 Pipe 对象传递给需要流的东西，但你可以传递它的 get 方法的返回值来做。
这对你有用吗？
【讨论】：
java.io.PipedOutputStream 和 java.io.PipedInputStream 看起来是用于此场景的类。它们旨在一起用于在线程之间传递数据。
如果你真的想要某个对象传递，它需要包含其中的每一个并通过 getter 公开它们。
【讨论】：
您不能创建同时派生自InputStream 和OutputStream 的类，因为它们不是接口并且它们具有通用方法并且Java 不允许多重继承（编译器不知道是否调用InputStream.close() 或 OutputStream.close() 如果您在新对象上调用 close()）。
另一个问题是缓冲区。 Java 想要为数据分配一个静态缓冲区（它不会改变）。这意味着当您使用 `java.io.PipedXxxStream' 时，写入数据最终会阻塞，除非您使用两个不同的线程。
所以来自 Apocalisp 的回答是正确的：你必须编写一个复制循环。
我建议您在项目中包含 Apache 的 commons-io，其中包含许多仅用于此类任务的辅助例程（在流、文件、字符串及其所有组合之间复制数据）。
【讨论】：
我认为这是很常见的事情。看到这个问题。
Easy way to write contents of a Java InputStream to an OutputStream
【讨论】：