PipedInputStream - Jak uniknąć "java.io.IOException: Pipe broken"

Question

Mam dwa wątki. Jeden z nich zapisuje do PipedOutputStream, inny odczytuje z odpowiedniego PipedInputStream. Tło polega na tym, że jeden wątek pobiera pewne dane ze zdalnego serwera i multipleksuje go do kilku innych wątków za pośrednictwem potokowanych strumieni.

Problem polega na tym, że czasami (szczególnie podczas pobierania dużego (> 50MB) pliki) otrzymuję java.io.IOException pipe uszkodzony podczas próby odczytu z PipedInputStream.
Javadoc mówi, że A pipe is said to be broken if a thread that was providing data bytes to the connected piped output stream is no longer alive.
To prawda, mój wątek w piśmie naprawdę umiera po zapisaniu wszystkich swoich danych do PipedOutputStream.

Jakieś rozwiązania? Jak mogę zapobiec wysyłaniu przez PipedInputStream tego wyjątku? Chcę móc odczytać wszystkie dane, które zostały napisane do PipedOutputStream, nawet jeśli pisanie wątku zakończyło jego pracę. (Jeśli ktoś wie, jak pisać wątek przy życiu, dopóki wszystkie dane nie zostaną odczytane, to rozwiązanie jest również dopuszczalne).

Answer 1

Użyj java.util.concurrent.CountDownLatch i nie kończ pierwszego wątku, zanim drugi sygnalizuje zakończenie odczytu z potoku.

Aktualizacja: szybki i brudny kod do zilustrowania mój komentarz poniżej

final PipedInputStream pin = getInputStream(); 
    final PipedOutputStream pout = getOutputStream(); 

    final CountDownLatch latch = new CountDownLatch(1); 

    InputStream in = new InputStream() { 

     @Override 
     public int read() throws IOException { 
      return pin.read(); 
     } 

     @Override 
     public void close() throws IOException { 
      super.close(); 
      latch.countDown(); 
     } 
    }; 


    OutputStream out = new OutputStream(){ 

     @Override 
     public void write(int b) throws IOException { 
      pout.write(b); 
     } 

     @Override 
     public void close() throws IOException { 
      while(latch.getCount()!=0) { 
       try { 
        latch.await(); 
       } catch (InterruptedException e) { 
        //too bad 
       } 
      } 
      super.close(); 
     } 
    }; 

    //give the streams to your threads, they don't know a latch ever existed 
    threadOne.feed(in); 
    threadTwo.feed(out); 

Answer 2

PipedInputStream i PipedOutputStream są podzielone (w odniesieniu do gwintowania). Zakładają, że każda instancja jest związana z konkretnym wątkiem. To dziwne. Sugeruję użycie własnej (lub przynajmniej innej) implementacji.

Answer 3

Czy zamykasz swoje PipedOutputStream, gdy wątek, który z niego korzysta, kończy się? Musisz to zrobić, aby bajty w nim zostały przepłukane do odpowiedniego PipedInputStream.