PipedInputStream/PipedOutputStream

PipedInputStream和PipedOutputStream分别是管道输入流和管道输出流.

它们的作用是让多线程之间可以通过管道进行通讯,在使用管道通信时,必须将PipedInputStream和PipedOutputStream配合使用.

使用管道通信时,大致流程是:线程A向PipedOutputStream中写入数据,这些数据会自动的发送到对应的pipedInputStream中进行缓存,此时,线程B通过读取PipedInputStream中的数据,就可以实现线程通信了.

实验一:发送简短的消息

Sender.java(发送消息)

public class Sender extends Thread {
    private PipedOutputStream pos = new PipedOutputStream();

    public Sender(PipedOutputStream pos) {
    this.pos = pos;
    }

    @Override
    public void run() {
         sendShortMessage();
    }

    // 发送简单的消息
    public void sendShortMessage() {
    try {
        pos.write("你好啊!".getBytes());
    } catch (IOException e) {
        e.printStackTrace();
    } finally {
        try {
        pos.close();
        } catch (IOException e) {
        e.printStackTrace();
        }
    }
    }
}

Reciver.java(接受消息)

public class Reciver extends Thread {
    private PipedInputStream pis = new PipedInputStream();

    public Reciver(PipedInputStream pis) {
    this.pis = pis;
    }

    @Override
    public void run() {
    byte[] buf = new byte[2048];
    try {
        pis.read(buf);
        pis.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
    System.out.println("Reciver :" + new String(buf));
    }
}

PipedTest.java

public class PipedTest {
    public static void main(String[] args) throws IOException {
    PipedOutputStream pipedOutputStream = new PipedOutputStream();
    PipedInputStream pipedInputStream = new PipedInputStream();
    Sender sender = new Sender(pipedOutputStream);
    Reciver reciver = new Reciver(pipedInputStream);
    pipedInputStream.connect(pipedOutputStream);
    sender.start();
    reciver.start();
    }
}
//Reciver :你好啊!

实验一很好理解:

pipedInputStream.connect(pipedOutputStream);方法把PipedOutPutStream和PipedInputStream关联了起来.
Sender通过write()方法,向PipedInputStream中写入"你好啊!",查看PipedOutputStream的writer()方法可知,其实是调用了PipedinputStream的receive()方法,继续查看该方法可知,PipedinputStream的

receive()方法,其实就是把数据保存到自己的byte buffer[]数组中,而且其数组的默认大小为1024.
Reciver通过read()方法,从自己的byte buffer[]数组中读取数据.

实验二:发送较长的消息

 Sender.java(发送消息)

public class Sender extends Thread {
    private PipedOutputStream pos = new PipedOutputStream();

    public Sender(PipedOutputStream pos) {
    this.pos = pos;
    }

    @Override
    public void run() {
    sendShortMessage();
    }

    // 发送较长的消息
    public void sendLongMessage() {
    StringBuilder sb = new StringBuilder();
    // 总共长度是1020个字节
    for (int i = 0; i < 102; i++) {
        sb.append("0123456789");
    }
    // 再写入26个字节。
    sb.append("abcdefghijklmnopqrstuvwxyz");
    // str的总长度是1020+26=1046个字节
    String str = sb.toString();
    try {
        pos.write(str.getBytes());
    } catch (IOException e) {
        e.printStackTrace();
    } finally {
        try {
        pos.close();
        } catch (IOException e) {
        e.printStackTrace();
        }
    }
    }
}

Reciver.java(接受消息)

public class Reciver extends Thread {
    private PipedInputStream pis = new PipedInputStream();

    public Reciver(PipedInputStream pis) {
    this.pis = pis;
    }

    @Override
    public void run() {
    byte[] buf = new byte[2048];
    try {
        pis.read(buf);
        pis.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
    System.out.println("Reciver :" + new String(buf));
    }
}

 PipedTest.java

public class PipedTest {
    public static void main(String[] args) throws IOException {
    PipedOutputStream pipedOutputStream = new PipedOutputStream();
    PipedInputStream pipedInputStream = new PipedInputStream();
    Sender sender = new Sender(pipedOutputStream);
    Reciver reciver = new Reciver(pipedInputStream);
    pipedInputStream.connect(pipedOutputStream);
    sender.start();
    reciver.start();
    }
}

 实验二的运行结果为:Reciver :012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789abcd

发现收到的数据少了"efghijklmnopqrstuvwxyz",为什么打印出来的数据正好是1024长度.其实上面我们已经说过了,PipedInputStream的byte [] buffer数组的默认长度只有1024,所以上面的输出是这样的.那怎么改才能输出全部内容的?

我们修改Reciver.java

public class Reciver extends Thread {
    private PipedInputStream pis = new PipedInputStream();

    public Reciver(PipedInputStream pis) {
    this.pis = pis;
    }

    @Override
    public void run() {
    reciverLongMessage();
    }

    public void reciverLongMessage() {
    while (true) {
        byte[] buf = new byte[1024];
        int len;
        try {
        len = pis.read(buf);
        if (len > 0) {
            System.out.println(new String(buf, 0, len));
        } else {
            break;
        }
        } catch (IOException e) {
        try {
            pis.close();
        } catch (IOException e2) {
            e2.printStackTrace();
        }
        }
    }
    }
}

 这样就可以正确输出结果啦!