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