Java并发(五)任务间使用管道进行通信

简介:

    通过I/O在线程间进行通信通常很有用。提供线程功能的类库以“管道”的形式对线程间的 I/O 提供了支持。它们在Java I/O 类库中的对应物就是PipedWriter(允许任务向管道写)和PipedReader(允许不同的任务从同一个管道中读取)。这个模型可以看做是“生产者-消费者”问题的变体,这里的管道就是一个封装好的解决方案。管道基本上是一个阻塞队列, 存在于多个引入BlockingQueue之前的Java版本中。

    下面是一个简单的例子,两个任务使用一个管道进行通信:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import  java.io.IOException;
import  java.io.PipedReader;
import  java.io.PipedWriter;
import  java.util.Random;
import  java.util.concurrent.ExecutorService;
import  java.util.concurrent.Executors;
import  java.util.concurrent.TimeUnit;
 
/**
  * 发送端
  */
class  Sender  implements  Runnable {
     private  Random rand =  new  Random( 47 );
     private  PipedWriter writer =  new  PipedWriter();
     public  PipedWriter getWriter() {  return  writer; }
     @Override
     public  void  run() {
         try  {
             while ( true ) {
                 for  ( char  c =  'A' ; c <  'z' ; c++) {
                     writer.write(c);
                     TimeUnit.MILLISECONDS.sleep(rand.nextInt( 500 ));
                 }
             }
         catch  (IOException e) {
             System.out.println(e +  " Sender write Exception" );
         catch  (InterruptedException e) {
             System.out.println(e +  " Sender sleep Interrupted" );
         }
     }
}
 
/**
  * 接收端
  */
class  Receiver  implements  Runnable {
     private  PipedReader reader;
     public  Receiver(Sender sender)  throws  IOException {
         reader =  new  PipedReader(sender.getWriter());
     }
     @Override
     public  void  run() {
         int  count =  0 ;
         try  {
             while ( true ) {
                 //在读取到内容之前,会一直阻塞
                 char  s = ( char )reader.read();
                 System.out.print( "Read: "  + s +  ", " );
                 if  (++count %  5  ==  0 ) {
                     System.out.println();
                 }
             }
         catch  (IOException e) {
             System.out.println(e +  " Receiver read Exception." );
         }
     }
}
 
public  class  PipedIO {
     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);
         TimeUnit.SECONDS.sleep( 5 );
         exec.shutdownNow();
     }
}

执行结果(可能的结果):

?
1
2
3
4
5
6
Read: A, Read: B, Read: C, Read: D, Read: E, 
Read: F, Read: G, Read: H, Read: I, Read: J, 
Read: K, Read: L, Read: M, Read: N, Read: O, 
Read: P, Read: Q, Read: R, Read: S, Read: T, 
Read: U, java.io.InterruptedIOException Receiver read Exception.
java.lang.InterruptedException: sleep interrupted Sender sleep Interrupted

    Sender和Receiver代表了需要互相通信的两个任务。Sender创建了一个PipedWriter,它是一个单独的对象;但是对于Receiver,PipedReader的建立必须在构造器中与一个PipedWriter相关联。就是说,PipedReader与PipedWriter的构造可以通过如下两种方式:

?
1
2
3
4
5
6
7
//方式一:先构造PipedReader,再通过它构造PipedWriter。
PipedReader reader =  new  PipedReader();
PipedWriter writer =  new  PipedWriter(reader);
 
//方式二:先构造PipedWriter,再通过它构造PipedReader。
PipedWriter writer2 =  new  PipedWriter();
PipedReader reader2 =  new  PipedReader(writer2);

    Sender把数据放进Writer,然后休眠一段时间(随机数)。然而,Receiver没有sleep()和wait。但当它调用read()时,如果没有更多的数据,管道将自动阻塞

    注意Sender和Receiver是在main()中启动的,即对象构造彻底完毕之后。如果你启动了一个没有构造完毕的对象,在不同的平台上管道可能会产生不一致的行为(注意,BlockingQueue使用起来更加健壮而容易)。

    在shutdownNow()被调用时,可以看到PipedReader与普通I/O之间最重要的差异——PipedReader是可以中断的。如果你将reader.read()替换为System.in.read(),那么interrupt()将不能打断read()调用。

目录
相关文章
|
6天前
|
监控 安全 算法
Java并发基础:LinkedTransferQueue全面解析!
LinkedTransferQueue类实现了高效的线程间数据传递,支持等待匹配的生产者-消费者模式,基于链表的无界设计使其在高并发场景下表现卓越,且无需担心队列溢出,丰富的方法和良好的可扩展性满足了各种复杂应用场景的需求。
Java并发基础:LinkedTransferQueue全面解析!
|
12天前
|
Java 程序员 API
Java并发基础:concurrent Flow API全面解析
java.util.concurrent.Flow定义了响应式编程的核心接口,促进了Java在异步数据处理和背压机制方面的标准化,这使得第三方库如Reactor和RxJava能够基于这些接口提供丰富的实现和功能,同时简化了响应式编程在Java中的使用,Flow API增强了Java在并发编程领域的灵活性,使得处理异步数据流变得更加自然和高效。
Java并发基础:concurrent Flow API全面解析
|
9天前
|
Java 流计算
这个错误信息来自于Java的ScheduledThreadPoolExecutor类,具体的原因是RejectedExecutionException,也就是任务被拒绝执行
【2月更文挑战第3天】这个错误信息来自于Java的ScheduledThreadPoolExecutor类,具体的原因是RejectedExecutionException,也就是任务被拒绝执行
22 2
|
6天前
|
存储 安全 Java
Java并发基础:BlockingQueue和BlockingDeque接口的区别?
BlockingQueue 和 BlockingDeque 它们都支持在并发编程中的线程安全操作,但是,这两个接口之间存在一些关键的区别,主要在于它们所支持的操作和数据结构的特性,
Java并发基础:BlockingQueue和BlockingDeque接口的区别?
|
6天前
|
Java 流计算
在Flink实时任务中,POJO(Plain Old Java Object)对象的模式演进可能会引起不兼容的问题
【2月更文挑战第6天】在Flink实时任务中,POJO(Plain Old Java Object)对象的模式演进可能会引起不兼容的问题
8 3
|
9天前
|
并行计算 Java API
深入理解Java多线程编程:创建、状态管理、同步与通信
深入理解Java多线程编程:创建、状态管理、同步与通信
|
9天前
|
缓存 安全 算法
Java并发基础:原子类之AtomicInteger全面解析
【2月更文挑战第2天】AtomicInteger类提供了线程安全的整数操作,它通过利用底层硬件的原子性指令,能够在多线程环境中高效地实现整数的无锁更新,避免了传统同步机制带来的性能开销,在高并发场景下成为计数器可大幅提高程序的执行效率,同时又保证了数据一致性。
56 15
Java并发基础:原子类之AtomicInteger全面解析
|
10天前
|
存储 算法 安全
Java并发基础:原子类之AtomicBoolean全面解析
【2月更文挑战第1天】 AtomicBoolean类优点在于能够确保布尔值在多线程环境下的原子性操作,避免了繁琐的同步措施,它提供了高效的非阻塞算法实现,可以大大提成程序的并发性能,AtomicBoolean的API设计非常简单易用。
Java并发基础:原子类之AtomicBoolean全面解析
|
10天前
|
数据采集 Java API
Java并发基础:CyclicBarrier全面解析!
CyclicBarrier的优点在于实现了线程间的相互等待与协同,确保所有线程在达到预定屏障点后才能继续执行,它支持屏障的重复使用,非常适合多轮次的任务同步,此外,CyclicBarrier还允许在屏障点执行特定操作,为复杂的多线程协作提供了便利。
Java并发基础:CyclicBarrier全面解析!
|
12天前
|
存储 安全 Java
Java并发基础:CopyOnWriteArraySet全面解析
CopyOnWriteArraySet类的优点在于能够实现无锁读取,确保高并发下的读取性能,同时,写操作通过复制底层数据来保证数据一致性,避免了多线程间的数据冲突,因此,它非常适合读多写少且对数据一致性要求较高的场景。
Java并发基础:CopyOnWriteArraySet全面解析

相关产品

  • 云迁移中心