Input-und Output-Stream-Leitung in Java

java.io.PipedOutputStream und java.io.PipedInputStream suchen, um die Klassen zu verwenden, die für dieses Szenario. Sie sind entworfen, um zusammen verwendet werden zum weiterleiten von Daten zwischen threads.

Wenn Sie wirklich wollen, einige einzelne Objekt zu übergeben, um es brauchen würde, um enthalten je eins von diesen und setzen Sie diese über Getter.

Können Sie nicht erstellen Sie eine Klasse, die sich sowohl aus InputStream und OutputStream denn diese sind nicht Schnittstellen, und Sie haben gemeinsame Methoden und Java erlaubt keine Mehrfachvererbung (der compiler nicht wissen, ob call InputStream.close() oder OutputStream.close() wenn Sie anrufen close() auf Ihrem neuen Objekt).

Das andere problem ist der Puffer. Java will, weisen Sie eine statische Puffer für die Daten (die ändert sich nicht). Dies bedeutet, wenn Sie die `java.io.PipedXxxStream', die das schreiben von Daten auf es wird schließlich blockieren, es sei denn, Sie verwenden zwei verschiedene threads.

Also die Antwort von Apocalisp ist richtig: Sie schreiben muss eine Kopie Schleife.

Schlage ich vor, dass Sie die Apache-commons-io in Ihr Projekt enthält viele hilfsroutinen, die nur für Aufgaben wie diese (kopieren von Daten zwischen streams, Dateien, Zeichenfolgen und alle Kombinationen davon).

Musste ich einen filter implementieren, der bei langsamen verbindungen zu Servlets also im Grunde wickelte ich die servlet-Ausgabestream in eine QueueOutputStream die jedes byte (in kleine Puffer), in eine Warteschlange, und dann Ausgang dieser kleine Puffer in einer 2. Ausgabe-stream, so in einer Weise, die diese Handlungen als input/output-stream, IMHO ist es sogar besser als JDK-Leitungen, die nicht skalieren, gut, im Grunde gibt es zu viel Kontext-Umschaltung in der standard-JDK-Implementierung (pro Lesen/schreiben), eine blockierende Warteschlange ist einfach perfekt für einen single-producer/consumer Szenario:

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;
              }
            }
    );
  }