Java中的输入和输出流管道

Input and Output Stream Pipe in Java

提问by Baginsss

Does anyone have any good suggestions for creating a Pipe object in Java which isboth an InputStream and and OutputStream since Java does not have multiple inheritance and both of the streams are abstract classes instead of interfaces?

有没有人有创建Java中的管道对象，任何好的建议是从Java既是一个InputStream和和OutputStream没有多重继承和两个流是抽象类，而不是接口？

The underlying need is to have a single object that can be passed to things which need either an InputStream or an OutputStream to pipe output from one thread to input for another.

潜在的需求是拥有一个可以传递给需要 InputStream 或 OutputStream 将输出从一个线程传输到另一个线程的输入的对象。

采纳答案by Lawrence Dol

It seems the point of this question is being missed. If I understand you correctly, you want an object that functions like an InputStream in one thread, and an OutputStream in another to create a means of communicating between the two threads.

似乎忽略了这个问题的重点。如果我理解正确的话，您希望一个对象在一个线程中像 InputStream 一样起作用，而在另一个线程中像一个 OutputStream 来创建两个线程之间的通信方式。

Perhaps one answer is to use composition instead of inheritance (which is recommended practice anyway). Create a Pipe which contains a PipedInputStream and a PipedOutputStream connected to each other, with getInputStream() and getOutputStream() methods.

也许一个答案是使用组合而不是继承（无论如何这是推荐的做法）。创建一个包含相互连接的 PipedInputStream 和 PipedOutputStream 的管道，使用 getInputStream() 和 getOutputStream() 方法。

You can't directly pass the Pipe object to something needing a stream, but you can pass the return value of it's get methods to do it.

您不能直接将 Pipe 对象传递给需要流的对象，但您可以传递它的 get 方法的返回值来执行此操作。

Does that work for you?

那对你有用吗？

回答by Apocalisp

This is a pretty common thing to do, I think. See this question.

我认为这是很常见的事情。看到这个问题。

Easy way to write contents of a Java InputStream to an OutputStream

将 Java InputStream 的内容写入 OutputStream 的简单方法

回答by mackenir

java.io.PipedOutputStream and java.io.PipedInputStream look to be the classes to use for this scenario. They are designed to be used together to pipe data between threads.

java.io.PipedOutputStream 和 java.io.PipedInputStream 看起来是用于此场景的类。它们被设计为一起使用以在线程之间传输数据。

If you really want some single object to pass around it would need to contain one of each of these and expose them via getters.

如果你真的想要一些单个对象传递，它需要包含其中的每一个并通过 getter 公开它们。

回答by Aaron Digulla

You can't create a class which derives both from InputStreamand OutputStreambecause these aren't interfaces and they have common methods and Java doesn't allow multiple inheritance (the compiler doesn't know whether to call InputStream.close()or OutputStream.close()if you call close()on your new object).

你不能创造出既从派生的类InputStream和OutputStream，因为这些都不是接口和他们有共同的方法和Java不允许多重继承（编译器不知道是否通话InputStream.close()还是OutputStream.close()如果你调用close()新的对象） .

The other problem is the buffer. Java wants to allocate a static buffer for the data (which doesn't change). This means when you use the `java.io.PipedXxxStream', the writing data to it will eventually block unless you use two different threads.

另一个问题是缓冲区。Java 想要为数据分配一个静态缓冲区（它不会改变）。这意味着当你使用 `java.io.PipedXxxStream' 时，写入数据最终会阻塞，除非你使用两个不同的线程。

So the answer from Apocalisp is correct: You must write a copy loop.

所以Apocalisp 的答案是正确的：你必须写一个复制循环。

I suggest that you include Apache's commons-io in your project which contains many helper routines just for tasks like this (copy data between streams, files, strings and all combinations thereof).

我建议您将 Apache 的 commons-io 包含在您的项目中，该项目包含许多仅用于此类任务的辅助例程（在流、文件、字符串及其所有组合之间复制数据）。

回答by Lawrence Dol

See http://ostermiller.org/utils/CircularBuffer.html

见http://ostermiller.org/utils/CircularBuffer.html

回答by Guido Medina

I had to implement a filter for slow connections to Servlets so basically I wrapped the servlet output stream into a QueueOutputStream which will add every byte (in small buffers), into a queue, and then output those small buffers to a 2nd output stream, so in a way this acts as input/output stream, IMHO this is better than JDK pipes which won't scale that well, basically there is too much context switching in the standard JDK implementation (per read/write), a blocking queue is just perfect for a single producer/consumer scenario:

我必须为到 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;
              }
            }
    );
  }