1-首先需要导入三个包

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;

2- 我们需要编写两个进行对话的线程类

2.1 - 发送消息的线程类 Sender

class Sender extends Thread{
    private PipedOutputStream out = new PipedOutputStream();
    public PipedOutputStream getOutputStream(){return out;}

    @Override
    public void run(){
        String s = new String("Receiver, 你好!");
        try {
            out.write(s.getBytes()); // 写入(发送)
            out.close();
        }catch (IOException e){
            System.out.println(e.getMessage());
        }
    }
}

2.2 - 接收消息的线程类 Receiver

class Receiver extends Thread{
    private PipedInputStream in = new PipedInputStream();
    public PipedInputStream getInputStream(){return in;}
    @Override
    public void run(){
        String s = null;
        byte[] buf = new byte[1024];
        try {
            int len = in.read(buf); // 读出信息
            s = new String(buf, 0, len);
            System.out.println("收到了以下信息: " + s);
            in.close();
        }catch (IOException e){
            System.out.println(e.getMessage());
        }
    }
}

声明的字节数组大小为1024原因是:类PipedInputStream运用的是一个1024字节固定大小的循环缓冲区。实际上，写入PipedOutputStream的数据保存到对应的PipedInputStream的内部缓冲区。

3 - 进行管道流测试的主类(管道流的使用)

public class PipeStreamDemo {
    public static void main(String[] args) {
        try {
            Sender sender = new Sender();   // 创建线程对象sender
            Receiver receiver = new Receiver(); // 创建线程对象receiver
            PipedOutputStream out = sender.getOutputStream(); // 写入
            PipedInputStream in = receiver.getInputStream(); // 读出
            out.connect(in); // 将输出发送到输入
            sender.start(); // 启动线程
            receiver.start();
        }catch (IOException e){
            System.out.println(e.getMessage());
        }
    }
}

4 - 输出示例

收到了以下信息: Receiver, 你好!