Java高并发之线程通信(八)---管道

浅浅的花香味﹌ 2022-05-05 17:26 223阅读 0赞

**Java高并发之线程通信(管道)**

Java管道原理：

广义上讲，管道就是一个用来在两个实体之间单项数据传输的导管。Java中的管道和linux中的管道是一样的，从本质上说，管道也是一种文件。

实际上原理很简单，如下图所示：

![70][]

管道流特点：

1. 管道数据流向的单向的，数据只能从一个进程(线程)流向另一个进程(线程)，如果要进行双向通信，必须建立两个管道。

2. 管道的读数据是一次性操作，数据一旦被读，它就从管道中被抛弃，释放空间以便写更多数据。

3. 当管道输出流write()导致管道缓冲区变满时，管道的write()调用将默认的被阻塞，等待缓冲区的数据被读取。同样的读进程也可能工作得比写进程块。当所有当前进程数据被读取时，管道变空。当这种情况发生时，一个随后的read()调用将默认被阻塞，等待缓冲区数据，这解决了read()调用返回文件结束的问题。

4. 管道输出流或者管道输入流的提前关闭，不会影响到对端流。比如输出流提前结束，输入流不会产生异常；输入流的提前结束也不会影响到输出流。

管道使用实例：

注意：管道输出流输出结束后一定要关闭输出流，因为只有关闭输出流之后，对端输入流read()才能返回null，如果没有关闭输出流会发生什么呢？

当输出流使用后关闭时：

> public class test \{  
>     public static class Write extends Thread\{  
>         public PipedOutputStream pos;  
>         Write(PipedOutputStream pos)\{  
>             this.pos = pos;  
>         \}  
>         public void run()\{  
>             PrintStream p = new PrintStream(pos);  
>             for(int i=1;i<10;i++)\{  
>                 p.println("hello");  
>                 p.flush();  
>             \}  
>             p.close();  
>         \}  
>     \}  
>       
>     public static class Read extends Thread\{  
>         public PipedInputStream pis;  
>         public String line = "null";  
>         Read(PipedInputStream pis)\{  
>             this.pis = pis;  
>         \}  
>         public void run()\{  
>             BufferedReader r = new BufferedReader(new InputStreamReader(pis));  
>             try \{  
>                 while(line!=null)\{  
>                     line = r.readLine();  
>                     System.out.println(getName()+": "+line);  
>                 \}  
>                 r.close();  
>             \} catch (IOException e) \{  
>                 e.printStackTrace();  
>             \}  
>         \}  
>     \}      
>    
>     public static void main(String args\[\]) throws InterruptedException, IOException\{  
>         //创建管道通信流  
>         PipedOutputStream pos = new PipedOutputStream();  
>         PipedInputStream pis = new PipedInputStream(pos);  
>         new Write(pos).start();  
>         new Read(pis).start();  
>     \}  
> \}

输出结果如下:

![70 1][]

可以看到最后管道输入流返回的是null。

当输出流使用完后不关闭时：

> public class test \{  
>     public static class Write extends Thread\{  
>         public PipedOutputStream pos;  
>         Write(PipedOutputStream pos)\{  
>             this.pos = pos;  
>         \}  
>         public void run()\{  
>             PrintStream p = new PrintStream(pos);  
>             for(int i=1;i<10;i++)\{  
>                 p.println("hello");  
>                 p.flush();  
>             \}  
>             //p.close();  
>         \}  
>     \}  
>       
>     public static class Read extends Thread\{  
>         public PipedInputStream pis;  
>         public String line = "null";  
>         Read(PipedInputStream pis)\{  
>             this.pis = pis;  
>         \}  
>         public void run()\{  
>             BufferedReader r = new BufferedReader(new InputStreamReader(pis));  
>             try \{  
>                 while(line!=null)\{  
>                     line = r.readLine();  
>                     System.out.println(getName()+": "+line);  
>                 \}  
>                 r.close();  
>             \} catch (IOException e) \{  
>                 e.printStackTrace();  
>             \}  
>         \}  
>     \}      
>    
>     public static void main(String args\[\]) throws InterruptedException, IOException\{  
>         //创建管道通信流  
>         PipedOutputStream pos = new PipedOutputStream();  
>         PipedInputStream pis = new PipedInputStream(pos);  
>         new Write(pos).start();  
>         new Read(pis).start();  
>     \}  
> \}

输出结果如下：

![70 2][]

可以看到输入流没有返回null，而是返回一个异常：Write end dead。

所以当管道通信结束后一定要关闭管道输出流。输入流可以通过read()方法是否返回null来判断管道通信是否结束。

工程中的管道使用：

上述的管道通信的例子只是用于实验的，在实际工程中我们并不会这样使用管道，把管道的两端(即管道输出流和管道输入流)设置为全局可见的是不安全的，会造成管道泄漏的问题。

我们看下面的代码：

>   
> public class test \{  
>     public static class Write extends Thread\{  
>         public PipedOutputStream pos;  
>         Write(PipedOutputStream pos)\{  
>             this.pos = pos;  
>         \}  
>         public void run()\{  
>             PrintStream p = new PrintStream(pos);  
>             for(int i=1;i<1000;i++)\{  
>                 p.println("hello");  
>                 p.flush();  
>             \}  
>             p.close();  
>         \}  
>     \}  
>       
>     public static class Read extends Thread\{  
>         public PipedInputStream pis;  
>         public String line = "null";  
>         Read(PipedInputStream pis)\{  
>             this.pis = pis;  
>         \}  
>         public void run()\{  
>             BufferedReader r = new BufferedReader(new InputStreamReader(pis));  
>             try \{  
>                 while(line!=null)\{  
>                     line = r.readLine();  
>                     System.out.println("read: "+line);  
>                 \}  
>                 r.close();  
>             \} catch (IOException e) \{  
>                 e.printStackTrace();  
>             \}  
>         \}  
>     \}  
>       
>     public static class Other\_Thread extends Thread\{  
>         public PipedInputStream pis;  
>         public String line = "null";  
>         Other\_Thread(PipedInputStream pis)\{  
>             this.pis = pis;  
>         \}  
>         public void run()\{  
>             BufferedReader r = new BufferedReader(new InputStreamReader(pis));  
>             try \{  
>                 while(line!=null)\{  
>                     line = r.readLine();  
>                     System.out.println("Other\_Thread: "+line);  
>                 \}  
>                 r.close();  
>             \} catch (IOException e) \{  
>                 e.printStackTrace();  
>             \}  
>         \}  
>     \}  
>    
>     public static void main(String args\[\]) throws InterruptedException, IOException\{  
>         //创建管道通信流  
>         PipedOutputStream pos = new PipedOutputStream();  
>         PipedInputStream pis = new PipedInputStream(pos);  
>         new Write(pos).start();  
>         new Read(pis).start();  
>         new Other\_Thread(pis).start();  
>     \}  
> \}

这里我多创建了一个线程Other\_Thread，用来模拟write线程和read线程正常通信的情况下，Other\_Thread线程获取了全局的管道输入流，并运行，看看会发生什么吧。

运行结果：

![70 3][]

这是运行结果的一个片段，但是已经可以说明问题了，write线程的管道数据被Other\_Thread线程截取了，这样的设计会产生管道泄漏问题。

在工程中更科学的利用管道应该如下:

在实际工程中一个线程往往需要跟多个线程通信，比如，write线程与read线程通信结束后，还需要write线程与其他线程通信，该怎么办呢？请看如下代码：

> public class test \{  
>     public static class Write extends Thread\{  
>         public PipedOutputStream pos = null;  
>           
>         //获取线程中的管道输出流  
>         public PipedOutputStream getPos()\{  
>             pos = new PipedOutputStream();  
>             return pos;  
>         \}  
>         //把数据通过管道输出流发送出去  
>         public void SentData()\{  
>             PrintStream p = new PrintStream(pos);  
>             for(int i=1;i<10;i++)\{  
>                 p.println("hello");  
>                 p.flush();  
>             \}  
>             p.close();  
>         \}  
>         @Override  
>         public void run()\{  
>             while(true);         //模拟耗时工作  
>         \}  
>     \}  
>       
>     public static class Read extends Thread\{  
>         public PipedInputStream pis = null;   
>         public String line = "null";  
>           
>         //获得线程中的管道输入流  
>         public PipedInputStream getPis()\{  
>             pis = new PipedInputStream();  
>             return pis;  
>         \}  
>         //利用管道输入流接收管道数据  
>         public void ReceiveData()\{  
>             BufferedReader r = new BufferedReader(new InputStreamReader(pis));  
>             try \{  
>                 while(line!=null)\{  
>                     line = r.readLine();  
>                     System.out.println("read: "+line);  
>                 \}  
>                 r.close();  
>             \} catch (IOException e) \{  
>                 e.printStackTrace();  
>             \}  
>         \}  
>         @Override  
>         public void run()\{  
>             while(true);        //模拟耗时工作  
>         \}  
>     \}  
>       
>     public static class Other\_Thread extends Thread\{  
>         public PipedInputStream pis = null;   
>         public String line = "null";  
>           
>         //获得线程中的管道输入流  
>         public PipedInputStream getPis()\{  
>             pis = new PipedInputStream();  
>             return pis;  
>         \}  
>         //利用管道输入流接收管道数据  
>         public void ReceiveData()\{  
>             BufferedReader r = new BufferedReader(new InputStreamReader(pis));  
>             try \{  
>                 while(line!=null)\{  
>                     line = r.readLine();  
>                     System.out.println("Other thread: "+line);  
>                 \}  
>                 r.close();  
>             \} catch (IOException e) \{  
>                 e.printStackTrace();  
>             \}  
>         \}  
>         @Override  
>         public void run()\{  
>             while(true);   //模拟耗时操作  
>         \}  
>     \}  
>     public static void main(String args\[\]) throws InterruptedException, IOException\{  
>         Write write = new Write();  
>         Read read = new Read();  
>         Other\_Thread other = new Other\_Thread();  
>         //连接两个线程的管道流 ---read和write线程  
>         write.getPos().connect(read.getPis());  
>         write.start();  
>         read.start();  
>         other.start();  
>         write.SentData();  
>         read.ReceiveData();  
>         Thread.sleep(2000);  
>         //重新连接两个线程的管道流 ---Other\_Thread和write线程  
>         write.getPos().connect(other.getPis());  
>         write.SentData();  
>         other.ReceiveData();  
>     \}  
> \}

运行结果：

![70 4][]

这只是输出的一个片段，但是已经能够说明问题了。

\---------------------

[70]: /images/20220505/4ea5afdc55394a8282be87dde63a0b59.png
[70 1]: /images/20220505/71d9d349bfae4ac3835d046e953f7d07.png
[70 2]: /images/20220505/323709855b84442b9018cd99a07f7515.png
[70 3]: /images/20220505/906324f19eeb4cf58eaa33d4b48be120.png
[70 4]: /images/20220505/aa4468fdbea749c7b4c29cf000ca8352.png