线程间通信管道流

不念不忘少年蓝@ 2022-09-11 01:19 192阅读 0赞

**管道输入/输出流**

管道流是是一种使用比较少的线程间通信方式，管道输入/输出流和普通文件输入/输出流或者网络输出/输出流不同之处在于，它主要用于线程之间的数据传输，传输的媒介为管道。

管道输入/输出流主要包括4种具体的实现：PipedOutputStrean、PipedInputStrean、PipedReader和PipedWriter，前两种面向字节，后两种面向字符。

java的管道的输入和输出实际上使用的是一个循环缓冲数组来实现的，默认为1024，输入流从这个数组中读取数据，输出流从这个数组中写入数据，当这个缓冲数组已满的时候，输出流所在的线程就会被阻塞，当向这个缓冲数组为空时，输入流所在的线程就会被阻塞。

管道是一个进程间通信的概念，在要通信的进程间构建一个**单向的数据流动**的通道。数据通过该通道从一个进程流向另一个进程时是**具有时间先后顺序**的。就像是在进程间架起了一个"管道"。

![image-20210910145315665][]

在java的类库中有2个管道类：PipedWriter(允许任务向管道写)，和PipedReader(允许不同任务从同一管道中读取)。管道也可以理解为一个缓冲区，将要读写的内容存入到管道，输入输出都要从这个管道去操作，管道提供了一个封装好的解决方案。

![image-20210910145329799][]

package com.dongguo.sync;
    
    
    import java.io.IOException;
    import java.io.PipedReader;
    import java.io.PipedWriter;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /** * @author Dongguo * @date 2021/8/24 0024-8:18 * @description: 任务间使用管道进行输入输出 */
    
    public class ThreadDemo6 { 
        public static void main(String[] args) throws Exception { 
            Sender sender = new Sender();
            Receiver receiver = new Receiver(sender);
            ExecutorService exec = Executors.newCachedThreadPool();
            exec.execute(sender);
            exec.execute(receiver);
            //休眠4秒钟后中断
            TimeUnit.SECONDS.sleep(4);
            exec.shutdownNow();
        }
    }
    
    class Sender implements Runnable { 
    
        private PipedWriter out = new PipedWriter();
    
        public PipedWriter getPipedWriter() { 
            return out;
        }
    
        public void run() { 
            try { 
                while (true) { 
                    for (char c = 'A'; c < 'Z'; c++) { 
                        out.write(c);
                        //随机休眠500以内毫秒
                        TimeUnit.MILLISECONDS.sleep(500);
                    }
                }
            } catch (IOException e) { 
                System.out.println(e + " Sender write exception");
            } catch (InterruptedException e) { 
                System.out.println(e + " Sender sleep interrupt");
            }
        }
    }
    
    class Receiver implements Runnable { 
        private PipedReader in;
    
        public Receiver(Sender sender) throws IOException { 
            in = new PipedReader(sender.getPipedWriter());
        }
    
        public void run() { 
            try { 
                while (true) { 
                    System.out.println("Read: " + (char) in.read() + ",");
                }
            } catch (IOException e) { 
                System.out.println(e + " Receiver read exception");
            }
        }
    }
    运行结果：
    Read: A,
    Read: B,
    Read: C,
    Read: D,
    Read: E,
    Read: F,
    java.io.InterruptedIOException Receiver read exception
    java.lang.InterruptedException: sleep interrupted Sender sleep interrupt

Sender和Receiver分别代表了2个需要互相通信的任务（相当于生产者和消费者的角色)，Sender 简历一个单独的PipedWriter对象，将信息写入到管道当中，而Receiver建立的PipedReader对象必须绑定到与Sender相关联的管道中，才能保证正确读取到数据。

[image-20210910145315665]: /images/20220829/b17ba935ac91404385c2851ebab0c9e4.png
[image-20210910145329799]: /images/20220829/57023dfce68542b89e2c3ab9873c355b.png