Java 使用pipe进行线程间通讯

Posted 2020-10-21 我也想学编程

Thinking in Java 中的例子，两个线程Sender&Receiver使用pipe进行通讯。Pipe是一个阻塞队列，解决了“生产者-消费者”线程通讯的问题。

 1 import java.io.IOException;
 2 import java.io.PipedReader;
 3 import java.io.PipedWriter;
 4 import java.util.concurrent.ExecutorService;
 5 import java.util.concurrent.Executors;
 6 
 7 class Sender implements Runnable
 8 {
 9     PipedWriter pipedWriter=new PipedWriter();
10     @Override public void run() {
11         try {
12             for (char a = ‘a‘; a <= ‘z‘; a++) {
13                 Thread.sleep(500);
14                 pipedWriter.write(a);
15                 //block
16             }
17         } catch (InterruptedException e) {
18             e.printStackTrace();
19         } catch (IOException e)
20         {
21             System.out.println("Close Pipe");
22         }
23     }
24     PipedWriter getPipedWriter()
25     {
26         return pipedWriter;
27     }
28 }
29 class Reciever implements Runnable
30 {
31     PipedReader pipedReader=null;
32     @Override public void run()
33     {
34         try{
35             while (true)
36             {
37                 char a=(char)pipedReader.read();
38                 System.out.print(a+", ");
39                 //block while nothing in the pipe
40             }
41         }catch (IOException e)
42         {
43             System.out.println("Close Pipe");
44         }
45     }
46     public Reciever(Sender sender) throws IOException
47     {//获得pipe
48         pipedReader=new PipedReader(sender.getPipedWriter());
49     }
50 }
51 public class Main{
52     public static void main(String[]args)
53     {
54         try {
55             Sender sender = new Sender();
56             Reciever reciever = new Reciever(sender);
57             ExecutorService service= Executors.newCachedThreadPool();
58             service.execute(sender);
59             service.execute(reciever);
60             Thread.sleep(20000);//跑一会
61             service.shutdown();//关闭
62             System.out.println("Over!");
63         }catch (IOException e)
64         {
65             e.printStackTrace();
66         }catch (InterruptedException e)
67         {
68             e.printStackTrace();
69         }
70 
71     }