Defectos con PipedInputStream/PipedOutputStream

Question

He visto dos respuestas en SO que afirman que las clases PipedInputStream y PipedOutputStream provistas por Java son defectuosas. Pero no dieron más detalles sobre lo que estaba mal con ellos. ¿Son realmente defectuosos, y si es así de qué manera? Actualmente estoy escribiendo un código que los usa, por lo que me gustaría saber si estoy tomando un giro equivocado.

One answer dijo:

PipedInputStream y PipedOutputStream están rotos (con respecto a enhebrar). Suponen que cada instancia está vinculada a un hilo en particular. Esto es extraño

Para mí eso no parece ni bizarro ni roto. Quizás el autor también tenía algunos otros defectos en mente?

Another answer dijo:

En la práctica, es mejor evitarlos. Los he usado una vez en 13 años y desearía no haberlo hecho.

Pero ese autor no pudo recordar cuál era el problema.

---

Al igual que con todas las clases, y especialmente a las clases utilizadas en varios subprocesos, que tendrán problemas si se les hace mal uso. Por lo tanto, no considero el impredecible "write end dead" IOException que PipedInputStream puede arrojar como un defecto (a falta de close() el PipedOutputStream conectado es un error; consulte el artículo Whats this? IOException: Write end dead, de Daniel Ferbers, para obtener más información). ¿Qué otros defectos reclamados hay?

Answer 1

No son defectuosos.

Al igual que con todas las clases, y especialmente las clases utilizadas en múltiples hilos, tendrá problemas si los usa mal. El impredecible "escribir final muerto" IOException que PipedInputStream puede arrojar no es un defecto (si falla close(), el PipedOutputStream conectado es un error; consulte el artículo Whats this? IOException: Write end dead, por Daniel Ferbers, para obtener más información).

Answer 2

Los he usado muy bien en mi proyecto y son invaluables para modificar flujos sobre la marcha y pasarlos. El único inconveniente parece ser que PipedInputStream tenía un corto buffer (alrededor de 1024) y mi outputtream estaba bombeando alrededor de 8KB.

No tiene ningún defecto y funciona perfectamente.

Ejemplo -------- en el maravilloso

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

Desde mi punto de vista hay un defecto. Más precisamente, existe un alto riesgo de un punto muerto si el subproceso que debe bombear datos en el PipedOutputStream muere prematuramente antes de que realmente escriba un solo byte en la secuencia. El problema en tal situación es que la implementación de las transmisiones por cañerías no es capaz de detectar la tubería rota. En consecuencia, la lectura de hilo de PipedInputStream esperará por siempre (es decirpunto muerto) en su primera llamada para leer().

La detección de tuberías rotas depende en realidad de la primera llamada a write() ya que la implementación inicializa la rosca del lado de escritura y solo a partir de ese momento la detección de tuberías rotas funcionará.

El código siguiente reproduce la situación:

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 4

Los defectos que veo con la implementación JDK son:

1) No hay tiempos de espera, lector o escritor puede bloquear infinitamente.

2) control subóptimo termina cuando se transfieren datos (se debe hacer solamente con ras, o cuando buffer circular está llena)

así que creé mi propia para abordar el (valor por encima, tiempo de espera pasó a través de un ThreadLocal) :

PipedOutputStream

Modo de empleo:

PiedOutputStreamTest

Espero que ayude ...