Недостатки с PipedInputStream / PipedOutputStream

Question

Я видел два ответа на SO, в которых утверждается, что классы PipedInputStream и PipedOutputStream, предоставляемые Java, имеют недостатки.Но они не уточнили, что с ними не так.Действительно ли они несовершенны, и если да, то каким образом?В настоящее время я пишу некоторый код, который использует их, поэтому я хотел бы знать, что я делаю неправильный ход.

Один ответ сказал:

PipedInputStream и PipedOutputStream сломаны (в отношении нарезания резьбы).Они предполагают, что каждый экземпляр связан с определенным потоком.Это странно.

Мне это не кажется ни странным, ни сломленным.Возможно, автор также имел в виду некоторые другие недостатки?

Другой ответ сказал:

На практике их лучше избегать.Я использовал их один раз в 13 лет, и мне жаль, что я этого не сделал.

Но этот автор не мог вспомнить, в чем проблема.

---

Как и во всех классахи особенно классы, используемые в нескольких потоках, у вас будут проблемы, если вы их неправильно используете.Поэтому я не считаю непредсказуемым «конец записи» IOException, который PipedInputStream может выбросить, как недостаток (ошибка close() подключенного PipedOutputStream является ошибкой; см. Статью Что это? IOException: Напишите end dead , автор Daniel Ferbers, для получения дополнительной информации).Какие еще заявленные недостатки есть?

Answer 1

Они не лишены недостатков.

Как и во всех классах, особенно в классах, используемых в нескольких потоках, у вас будут проблемы, если вы их неправильно используете.Непредсказуемый «конец записи» IOException, который может выдать PipedInputStream, не является недостатком (ошибка close() подключенного PipedOutputStream является ошибкой; см. Статью Что это? IOException: Writeend dead , Даниэль Ферберс, для получения дополнительной информации).

Answer 2

Я хорошо использовал их в своем проекте, и они неоценимы для изменения потоков на лету и их передачи. Единственным недостатком, по-видимому, было то, что PipedInputStream имел короткий буфер (около 1024), а мой выходной поток загружал около 8 КБ.

У него нет дефектов, и он прекрасно работает.

-------- Пример в Groovy

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

Один недостаток может заключаться в том, что автор не может четко указать читателю, что он столкнулся с проблемой:

PipedOutputStream out = new PipedOutputStream();
PipedInputStream in = new PipedInputStream(out);

new Thread(() -> {
    try {
        writeToOut(out);
        out.close();
    }
    catch (SomeDataProviderException e) {
        // Have to notify the reading side, but how?
    }
}).start();

readFromIn(in);

Автор может закрыть out,но, возможно, читатель неверно истолковывает это как конец данных.Для правильной обработки необходима дополнительная логика.Было бы проще, если бы была предусмотрена функция для разрыва трубы вручную.

Теперь есть JDK-8222924 , который запрашивает способ разрыва трубы вручную.

Answer 4

Недостатки, которые я вижу в реализации JDK:

1) Ни один таймаут, читатель или писатель не может блокировать бесконечно.

2) Неоптимальное управление при передаче данных (должно выполняться только при очистке или при заполнении циклического буфера)

Итак, я создал свой собственный для решения вышеуказанного, (значение тайм-аута передается через ThreadLocal):

PipedOutputStream

Как использовать:

PiedOutputStreamTest

Надеюсь, это поможет ...

Answer 5

С моей точки зрения есть недостаток. Точнее, существует высокий риск тупиковой ситуации, если Поток, который должен закачать данные в PipedOutputStream, преждевременно умирает, прежде чем он фактически записывает один поток в поток. Проблема в такой ситуации заключается в том, что реализация потоковых каналов не способна обнаружить сломанный канал. Следовательно, чтение потока из PipedInputStream будет ждать вечно (т.е. тупик) при первом вызове read ().

Обнаружение разорванных каналов фактически зависит от первого вызова write (), поскольку реализация будет лениво инициализировать поток на стороне записи и только с этого момента будет работать обнаружение разорванных каналов.

Следующий код воспроизводит ситуацию:

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