java - 如何通知 PipedInputStream 线程 PipedOutputStream 线程已写入最后一个字节？

Question

如何正确完成管道输出端的工作？我需要写入线程终止或做一些其他工作，而读取线程读取所有写入数据直到结束。

我应该在写入端关闭管道还是什么？

更新 1

我想澄清一下......根据给定的答案，我认为设计管道行为不假设任何优雅终止是否正确？

即一旦打开，停止管道的唯一方法就是打破管道？

当 read() 方法返回 -1 时，传统流期望流信号结束。认为管道流永远不会发生这种情况是正确的吗？

Answer 1

是的，关闭 PipedOutputStream 会导致 PipedInputStream 出现 -1。

我觉得很优雅!这是我的 SSCCE :

import java.io.*;
import java.nio.charset.*;

public class SOPipe
{
    public static void main(String[] args) throws Exception
    {
        PipedOutputStream os = new PipedOutputStream();
        PipedInputStream is = new PipedInputStream(os);

        ReaderThread readerThread = new ReaderThread(is);
        WriterThread writerThread = new WriterThread(os);

        readerThread.start();
        writerThread.start();

        readerThread.join();
        writerThread.join();

        System.out.println("Both Reader and Writer completed.");
        System.out.println("Main method returning normally now.");
    }

    private static final Charset LATIN1 = Charset.forName("latin1");

    public static class WriterThread extends Thread
    {
        private final PipedOutputStream _os;

        public WriterThread(PipedOutputStream os)
        {
            _os = os;
        }

        public void run()
        {
            try
            {
                String msg = "Ceci n'est pas une pipe";
                byte[] msgBytes = msg.getBytes(LATIN1);
                System.out.println("WriterThread sending message: " + msg);
                for(int i = 0; i < msgBytes.length; i++)
                {
                    _os.write(msgBytes, i, 1);
                    System.out.println("WriterThread wrote a byte!");
                    _os.flush();
                }
                _os.close();
                System.out.println("[COMPLETED] WriterThread");
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }
    }

    public static class ReaderThread extends Thread
    {
        private final PipedInputStream _is;

        public ReaderThread(PipedInputStream is)
        {
            _is = is;
        }

        public void run()
        {
            try
            {
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                byte[] buffer = new byte[1];
                int read;
                while ((read = _is.read(buffer, 0, 1)) != -1)
                {
                    System.out.println("ReaderThread read a byte!");
                    baos.write(buffer, 0, read);
                }
                System.out.println("[COMPLETED] ReaderThread; received: " 
                        + new String(baos.toByteArray(), LATIN1));
                _is.close();
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }
    }
}