【并发编程】PipedOutputStream和PipedInputStream管道实现线程通信

ゝ一世哀愁。 2022-03-19 06:50 182阅读 0赞

# 前言 #

Github：[https://github.com/yihonglei/jdk-source-code-reading][https_github.com_yihonglei_jdk-source-code-reading]（java-concurrent）

# 一 PipedOutputStream和PipedInputStream简介 #

PipedOutputStream和PipedInputStream通过字节流实现线程间的通信，

通过PipedOutputStream.connect(PipedInputStream snk)或PipedInputStream.connect(PipedOutputStream src)建立连接。

# 二 代码实例 #

1、使用PipedOutputStream输出流写入数据

package com.jpeony.concurrent.piped;
    
    import java.io.IOException;
    import java.io.PipedOutputStream;
    
    /**
     * 写入数据
     *
     * @author yihonglei
     */
    public class WriteData {
        public void writeMethod(PipedOutputStream out) throws IOException {
            try {
                String outData = "I love you";
                out.write(outData.getBytes());
                System.out.println("写入数据:" + outData);
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                // 关闭输出流
                out.close();
            }
        }
    }

2、使用PipedInputStream输入流读出数据

package com.jpeony.concurrent.piped;
    
    import java.io.IOException;
    import java.io.PipedInputStream;
    
    /**
     * 读出数据
     *
     * @author yihonglei
     */
    public class ReadData {
        public void readMethod(PipedInputStream input) throws IOException {
            try {
                byte[] byteArray = new byte[20];
                int readLength = input.read(byteArray);
                while (readLength != -1) {
                    String newData = new String(byteArray, 0, readLength);
                    System.out.println("读出数据:" + newData);
                    readLength = input.read(byteArray);
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                // 关闭流
                input.close();
            }
        }
    }

3、写入数据线程

package com.jpeony.concurrent.piped;
    
    import java.io.IOException;
    import java.io.PipedOutputStream;
    
    /**
     * 写入数据线程
     *
     * @author yihonglei
     */
    public class WriteThread extends Thread {
        private WriteData writeData;
        private PipedOutputStream out;
    
        public WriteThread(WriteData writeData, PipedOutputStream out) {
            this.writeData = writeData;
            this.out = out;
        }
    
        @Override
        public void run() {
            try {
                writeData.writeMethod(out);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

4、读出数据线程

package com.jpeony.concurrent.piped;
    
    import java.io.IOException;
    import java.io.PipedInputStream;
    
    /**
     * 读出数据线程
     *
     * @author yihonglei
     */
    public class ReadThread extends Thread {
        private ReadData readData;
        private PipedInputStream input;
    
        public ReadThread(ReadData readData, PipedInputStream input) {
            this.readData = readData;
            this.input = input;
        }
    
        @Override
        public void run() {
            try {
                readData.readMethod(input);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

5、测试类

package com.jpeony.concurrent.piped;
    
    import java.io.IOException;
    import java.io.PipedInputStream;
    import java.io.PipedOutputStream;
    
    /**
     * 测试类
     *
     * @author yihonglei
     */
    public class RunTest {
        public static void main(String[] args) {
            try {
                WriteData writeData = new WriteData();
                ReadData readData = new ReadData();
    
                PipedInputStream inputStream = new PipedInputStream();
                PipedOutputStream outputStream = new PipedOutputStream();
    
                /*
                 * 通过 inputStream.connect(outputStream)
                 * 或
                 * outputStream.connect(inputStream)
                 * 使得两个Stream之间产生通信链接，进行数据交换输出与输入。
                 */
                // inputStream.connect(outputStream);
                outputStream.connect(inputStream);
    
                ReadThread readThread = new ReadThread(readData, inputStream);
                readThread.start();
    
                // 休眠1秒
                Thread.sleep(1000);
    
                WriteThread writeThread = new WriteThread(writeData, outputStream);
                writeThread.start();
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

6、运行结果

![20190215174336720.jpg][]

7、代码分析

1）代码中通过 inputStream.connect(outputStream)或outputStream.connect(inputStream)

使得两个Stream之间产生通信链接，进行数据交换输出与输入，从而实现线程通信。

(2）当线程ReadThread启动时，int readLength = input.read(byteArray)未读取到数据，

只有当线程WrithThread启动后，开始写入数据，ReadThread才能够读取到数据。

[https_github.com_yihonglei_jdk-source-code-reading]: https://github.com/yihonglei/jdk-source-code-reading
[20190215174336720.jpg]: /images/20220319/f1a91c85dce74d809351543d00a17069.png