Java并发(五)任务间使用管道进行通信-阿里云开发者社区

开发者社区> 小龙猫> 正文

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()调用。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
spring-boot | 多线程并发定时任务
刚刚看了下Spring Boot实现定时任务的文章,感觉还不错。Spring Boot 使用Spring自带的Schedule来实现定时任务变得非常简单和方便。在这里个大家分享下。
3048 0
java 计划任务
有一些程序是永不停歇的,比如监控某个数据表中的内容是否变化。若有变化通知给其他程序。下面两个类可以作为模版。
648 0
使用OpenApi弹性释放和设置云服务器ECS释放
云服务器ECS的一个重要特性就是按需创建资源。您可以在业务高峰期按需弹性的自定义规则进行资源创建,在完成业务计算的时候释放资源。本篇将提供几个Tips帮助您更加容易和自动化的完成云服务器的释放和弹性设置。
8268 0
麻省理工学院开发系统 使用肌肉信号控制无人机
麻省理工学院的研究人员创造了一个新的系统,能让我们更接近于人与机器人的无缝协作。这套系统被称为 "行为-A-Bot"。它利用可穿戴式传感器发出的人体肌肉信号来驾驶无人机,控制无人机的运动。
357 0
+关注
125
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
文娱运维技术
立即下载
《SaaS模式云原生数据仓库应用场景实践》
立即下载
《看见新力量:二》电子书
立即下载