java 管道介紹
在java中，PipedOutputStream和PipedInputStream分別是管道輸出流和管道輸入流。
它們的作用是讓多線程可以通過管道進行線程間的通訊。在使用管道通信時，必須將PipedOutputStream和PipedInputStream配套使用。
使用管道通信時，大致的流程是：我們在線程A中向PipedOutputStream中寫入數據，這些數據會自動的發送到與PipedOutputStream對應的PipedInputStream中，進而存儲在PipedInputStream的緩沖中；此時，線程B通過讀取PipedInputStream中的數據。就可以實現，線程A和線程B的通信。
PipedOutputStream和PipedInputStream源碼分析
下面介紹PipedOutputStream和PipedInputStream的源碼。在閱讀它們的源碼之前，建議先看看源碼后面的示例。待理解管道的作用和用法之后，再看源碼，可能更容易理解。
1. PipedOutputStream 源碼分析(基于jdk1.7.40)
|
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
package java.io;import java.io.*;public class PipedOutputStream extends OutputStream { // 與PipedOutputStream通信的PipedInputStream對象 private PipedInputStream sink; // 構造函數，指定配對的PipedInputStream public PipedOutputStream(PipedInputStream snk) throws IOException { connect(snk); } // 構造函數 public PipedOutputStream() { } // 將“管道輸出流” 和 “管道輸入流”連接。 public synchronized void connect(PipedInputStream snk) throws IOException { if (snk == null) { throw new NullPointerException(); } else if (sink != null || snk.connected) { throw new IOException("Already connected"); } // 設置“管道輸入流” sink = snk; // 初始化“管道輸入流”的讀寫位置 // int是PipedInputStream中定義的，代表“管道輸入流”的讀寫位置 snk.in = -1; // 初始化“管道輸出流”的讀寫位置。 // out是PipedInputStream中定義的，代表“管道輸出流”的讀寫位置 snk.out = 0; // 設置“管道輸入流”和“管道輸出流”為已連接狀態 // connected是PipedInputStream中定義的，用于表示“管道輸入流與管道輸出流”是否已經連接 snk.connected = true; } // 將int類型b寫入“管道輸出流”中。 // 將b寫入“管道輸出流”之后，它會將b傳輸給“管道輸入流” public void write(int b) throws IOException { if (sink == null) { throw new IOException("Pipe not connected"); } sink.receive(b); } // 將字節數組b寫入“管道輸出流”中。 // 將數組b寫入“管道輸出流”之后，它會將其傳輸給“管道輸入流” public void write(byte b[], int off, int len) throws IOException { if (sink == null) { throw new IOException("Pipe not connected"); } else if (b == null) { throw new NullPointerException(); } else if ((off < 0) || (off > b.length) || (len < 0) || ((off + len) > b.length) || ((off + len) < 0)) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return; } // “管道輸入流”接收數據 sink.receive(b, off, len); } // 清空“管道輸出流”。 // 這里會調用“管道輸入流”的notifyAll()； // 目的是讓“管道輸入流”放棄對當前資源的占有，讓其它的等待線程(等待讀取管道輸出流的線程)讀取“管道輸出流”的值。 public synchronized void flush() throws IOException { if (sink != null) { synchronized (sink) { sink.notifyAll(); } } } // 關閉“管道輸出流”。 // 關閉之后，會調用receivedLast()通知“管道輸入流”它已經關閉。 public void close() throws IOException { if (sink != null) { sink.receivedLast(); } }} |
2. PipedInputStream 源碼分析(基于jdk1.7.40)
|
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
|
package java.io;public class PipedInputStream extends InputStream { // “管道輸出流”是否關閉的標記 boolean closedByWriter = false; // “管道輸入流”是否關閉的標記 volatile boolean closedByReader = false; // “管道輸入流”與“管道輸出流”是否連接的標記 // 它在PipedOutputStream的connect()連接函數中被設置為true boolean connected = false; Thread readSide; // 讀取“管道”數據的線程 Thread writeSide; // 向“管道”寫入數據的線程 // “管道”的默認大小 private static final int DEFAULT_PIPE_SIZE = 1024; protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE; // 緩沖區 protected byte buffer[]; //下一個寫入字節的位置。in==out代表滿，說明“寫入的數據”全部被讀取了。 protected int in = -1; //下一個讀取字節的位置。in==out代表滿，說明“寫入的數據”全部被讀取了。 protected int out = 0; // 構造函數：指定與“管道輸入流”關聯的“管道輸出流” public PipedInputStream(PipedOutputStream src) throws IOException { this(src, DEFAULT_PIPE_SIZE); } // 構造函數：指定與“管道輸入流”關聯的“管道輸出流”，以及“緩沖區大小” public PipedInputStream(PipedOutputStream src, int pipeSize) throws IOException { initPipe(pipeSize); connect(src); } // 構造函數：默認緩沖區大小是1024字節 public PipedInputStream() { initPipe(DEFAULT_PIPE_SIZE); } // 構造函數：指定緩沖區大小是pipeSize public PipedInputStream(int pipeSize) { initPipe(pipeSize); } // 初始化“管道”：新建緩沖區大小 private void initPipe(int pipeSize) { if (pipeSize <= 0) { throw new IllegalArgumentException("Pipe Size <= 0"); } buffer = new byte[pipeSize]; } // 將“管道輸入流”和“管道輸出流”綁定。 // 實際上，這里調用的是PipedOutputStream的connect()函數 public void connect(PipedOutputStream src) throws IOException { src.connect(this); } // 接收int類型的數據b。 // 它只會在PipedOutputStream的write(int b)中會被調用 protected synchronized void receive(int b) throws IOException { // 檢查管道狀態 checkStateForReceive(); // 獲取“寫入管道”的線程 writeSide = Thread.currentThread(); // 若“寫入管道”的數據正好全部被讀取完，則等待。 if (in == out) awaitSpace(); if (in < 0) { in = 0; out = 0; } // 將b保存到緩沖區 buffer[in++] = (byte)(b & 0xFF); if (in >= buffer.length) { in = 0; } } // 接收字節數組b。 synchronized void receive(byte b[], int off, int len) throws IOException { // 檢查管道狀態 checkStateForReceive(); // 獲取“寫入管道”的線程 writeSide = Thread.currentThread(); int bytesToTransfer = len; while (bytesToTransfer > 0) { // 若“寫入管道”的數據正好全部被讀取完，則等待。 if (in == out) awaitSpace(); int nextTransferAmount = 0; // 如果“管道中被讀取的數據，少于寫入管道的數據”； // 則設置nextTransferAmount=“buffer.length - in” if (out < in) { nextTransferAmount = buffer.length - in; } else if (in < out) { // 如果“管道中被讀取的數據，大于/等于寫入管道的數據”，則執行后面的操作 // 若in==-1(即管道的寫入數據等于被讀取數據)，此時nextTransferAmount = buffer.length - in; // 否則，nextTransferAmount = out - in; if (in == -1) { in = out = 0; nextTransferAmount = buffer.length - in; } else { nextTransferAmount = out - in; } } if (nextTransferAmount > bytesToTransfer) nextTransferAmount = bytesToTransfer; // assert斷言的作用是，若nextTransferAmount <= 0，則終止程序。 assert(nextTransferAmount > 0); // 將數據寫入到緩沖中 System.arraycopy(b, off, buffer, in, nextTransferAmount); bytesToTransfer -= nextTransferAmount; off += nextTransferAmount; in += nextTransferAmount; if (in >= buffer.length) { in = 0; } } } // 檢查管道狀態 private void checkStateForReceive() throws IOException { if (!connected) { throw new IOException("Pipe not connected"); } else if (closedByWriter || closedByReader) { throw new IOException("Pipe closed"); } else if (readSide != null && !readSide.isAlive()) { throw new IOException("Read end dead"); } } // 等待。 // 若“寫入管道”的數據正好全部被讀取完(例如，管道緩沖滿)，則執行awaitSpace()操作； // 它的目的是讓“讀取管道的線程”管道產生讀取數據請求，從而才能繼續的向“管道”中寫入數據。 private void awaitSpace() throws IOException { // 如果“管道中被讀取的數據，等于寫入管道的數據”時， // 則每隔1000ms檢查“管道狀態”，并喚醒管道操作：若有“讀取管道數據線程被阻塞”，則喚醒該線程。 while (in == out) { checkStateForReceive(); /* full: kick any waiting readers */ notifyAll(); try { wait(1000); } catch (InterruptedException ex) { throw new java.io.InterruptedIOException(); } } } // 當PipedOutputStream被關閉時，被調用 synchronized void receivedLast() { closedByWriter = true; notifyAll(); } // 從管道(的緩沖)中讀取一個字節，并將其轉換成int類型 public synchronized int read() throws IOException { if (!connected) { throw new IOException("Pipe not connected"); } else if (closedByReader) { throw new IOException("Pipe closed"); } else if (writeSide != null && !writeSide.isAlive() && !closedByWriter && (in < 0)) { throw new IOException("Write end dead"); } readSide = Thread.currentThread(); int trials = 2; while (in < 0) { if (closedByWriter) { /* closed by writer, return EOF */ return -1; } if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) { throw new IOException("Pipe broken"); } /* might be a writer waiting */ notifyAll(); try { wait(1000); } catch (InterruptedException ex) { throw new java.io.InterruptedIOException(); } } int ret = buffer[out++] & 0xFF; if (out >= buffer.length) { out = 0; } if (in == out) { /* now empty */ in = -1; } return ret; } // 從管道(的緩沖)中讀取數據，并將其存入到數組b中 public synchronized int read(byte b[], int off, int len) throws IOException { if (b == null) { throw new NullPointerException(); } else if (off < 0 || len < 0 || len > b.length - off) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return 0; } /* possibly wait on the first character */ int c = read(); if (c < 0) { return -1; } b[off] = (byte) c; int rlen = 1; while ((in >= 0) && (len > 1)) { int available; if (in > out) { available = Math.min((buffer.length - out), (in - out)); } else { available = buffer.length - out; } // A byte is read beforehand outside the loop if (available > (len - 1)) { available = len - 1; } System.arraycopy(buffer, out, b, off + rlen, available); out += available; rlen += available; len -= available; if (out >= buffer.length) { out = 0; } if (in == out) { /* now empty */ in = -1; } } return rlen; } // 返回不受阻塞地從此輸入流中讀取的字節數。 public synchronized int available() throws IOException { if(in < 0) return 0; else if(in == out) return buffer.length; else if (in > out) return in - out; else return in + buffer.length - out; } // 關閉管道輸入流 public void close() throws IOException { closedByReader = true; synchronized (this) { in = -1; } }} |
管道通信示例
下面，我們看看多線程中通過管道通信的例子。例子中包括3個類：Receiver.java, PipedStreamTest.java 和 Sender.java。
Receiver.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
|
import java.io.IOException; import java.io.PipedInputStream; @SuppressWarnings("all") /** * 接收者線程 */public class Receiver extends Thread { // 管道輸入流對象。 // 它和“管道輸出流(PipedOutputStream)”對象綁定， // 從而可以接收“管道輸出流”的數據，再讓用戶讀取。 private PipedInputStream in = new PipedInputStream(); // 獲得“管道輸入流”對象 public PipedInputStream getInputStream(){ return in; } @Override public void run(){ readMessageOnce() ; //readMessageContinued() ; } // 從“管道輸入流”中讀取1次數據 public void readMessageOnce(){ // 雖然buf的大小是2048個字節，但最多只會從“管道輸入流”中讀取1024個字節。 // 因為，“管道輸入流”的緩沖區大小默認只有1024個字節。 byte[] buf = new byte[2048]; try { int len = in.read(buf); System.out.println(new String(buf,0,len)); in.close(); } catch (IOException e) { e.printStackTrace(); } } // 從“管道輸入流”讀取>1024個字節時，就停止讀取 public void readMessageContinued() { int total=0; while(true) { byte[] buf = new byte[1024]; try { int len = in.read(buf); total += len; System.out.println(new String(buf,0,len)); // 若讀取的字節總數>1024，則退出循環。 if (total > 1024) break; } catch (IOException e) { e.printStackTrace(); } } try { in.close(); } catch (IOException e) { e.printStackTrace(); } }} |
Sender.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
|
import java.io.IOException; import java.io.PipedOutputStream; @SuppressWarnings("all")/** * 發送者線程 */public class Sender extends Thread { // 管道輸出流對象。 // 它和“管道輸入流(PipedInputStream)”對象綁定， // 從而可以將數據發送給“管道輸入流”的數據，然后用戶可以從“管道輸入流”讀取數據。 private PipedOutputStream out = new PipedOutputStream(); // 獲得“管道輸出流”對象 public PipedOutputStream getOutputStream(){ return out; } @Override public void run(){ writeShortMessage(); //writeLongMessage(); } // 向“管道輸出流”中寫入一則較簡短的消息："this is a short message" private void writeShortMessage() { String strInfo = "this is a short message" ; try { out.write(strInfo.getBytes()); out.close(); } catch (IOException e) { e.printStackTrace(); } } // 向“管道輸出流”中寫入一則較長的消息 private void writeLongMessage() { StringBuilder sb = new StringBuilder(); // 通過for循環寫入1020個字節 for (int i=0; i<102; i++) sb.append("0123456789"); // 再寫入26個字節。 sb.append("abcdefghijklmnopqrstuvwxyz"); // str的總長度是1020+26=1046個字節 String str = sb.toString(); try { // 將1046個字節寫入到“管道輸出流”中 out.write(str.getBytes()); out.close(); } catch (IOException e) { e.printStackTrace(); } }} |
PipedStreamTest.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
|
import java.io.PipedInputStream;import java.io.PipedOutputStream;import java.io.IOException;@SuppressWarnings("all") /** * 管道輸入流和管道輸出流的交互程序 */public class PipedStreamTest { public static void main(String[] args) { Sender t1 = new Sender(); Receiver t2 = new Receiver(); PipedOutputStream out = t1.getOutputStream(); PipedInputStream in = t2.getInputStream(); try { //管道連接。下面2句話的本質是一樣。 //out.connect(in); in.connect(out); /** * Thread類的START方法： * 使該線程開始執行；Java 虛擬機調用該線程的 run 方法。 * 結果是兩個線程并發地運行；當前線程（從調用返回給 start 方法）和另一個線程（執行其 run 方法）。 * 多次啟動一個線程是非法的。特別是當線程已經結束執行后，不能再重新啟動。 */ t1.start(); t2.start(); } catch (IOException e) { e.printStackTrace(); } }} |
運行結果：
this is a short message
說明：
(01)
in.connect(out);
將“管道輸入流”和“管道輸出流”關聯起來。查看PipedOutputStream.java和PipedInputStream.java中connect()的源碼；我們知道 out.connect(in); 等價于 in.connect(out);
(02)
t1.start(); // 啟動“Sender”線程
t2.start(); // 啟動“Receiver”線程
先查看Sender.java的源碼，線程啟動后執行run()函數；在Sender.java的run()中，調用writeShortMessage();
writeShortMessage();的作用就是向“管道輸出流”中寫入數據"this is a short message" ；這條數據會被“管道輸入流”接收到。下面看看這是如何實現的。
先看write(byte b[])的源碼，在OutputStream.java中定義。PipedOutputStream.java繼承于OutputStream.java；OutputStream.java中write(byte b[])的源碼如下：
|
1
2
3
|
public void write(byte b[]) throws IOException { write(b, 0, b.length);} |
實際上write(byte b[])是調用的PipedOutputStream.java中的write(byte b[], int off, int len)函數。查看write(byte b[], int off, int len)的源碼，我們發現：它會調用 sink.receive(b, off, len); 進一步查看receive(byte b[], int off, int len)的定義，我們知道sink.receive(b, off, len)的作用就是：將“管道輸出流”中的數據保存到“管道輸入流”的緩沖中。而“管道輸入流”的緩沖區buffer的默認大小是1024個字節。
至此，我們知道：t1.start()啟動Sender線程，而Sender線程會將數據"this is a short message"寫入到“管道輸出流”；而“管道輸出流”又會將該數據傳輸給“管道輸入流”，即而保存在“管道輸入流”的緩沖中。
接下來，我們看看“用戶如何從‘管道輸入流'的緩沖中讀取數據”。這實際上就是Receiver線程的動作。
t2.start() 會啟動Receiver線程，從而執行Receiver.java的run()函數。查看Receiver.java的源碼，我們知道run()調用了readMessageOnce()。
而readMessageOnce()就是調用in.read(buf)從“管道輸入流in”中讀取數據，并保存到buf中。
通過上面的分析，我們已經知道“管道輸入流in”的緩沖中的數據是"this is a short message"；因此，buf的數據就是"this is a short message"。
為了加深對管道的理解。我們接著進行下面兩個小試驗。
試驗一：修改Sender.java
將
|
1
2
3
4
|
public void run(){ writeShortMessage(); //writeLongMessage();} |
修改為
|
1
2
3
4
|
public void run(){ //writeShortMessage(); writeLongMessage();} |
運行程序。運行結果為：
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
012345678901234567890123456789abcd
這些數據是通過writeLongMessage()寫入到“管道輸出流”，然后傳送給“管道輸入流”，進而存儲在“管道輸入流”的緩沖中；再被用戶從緩沖讀取出來的數據。
然后，觀察writeLongMessage()的源碼。我們可以發現，str的長度是1046個字節，然后運行結果只有1024個字節！為什么會這樣呢？
道理很簡單：管道輸入流的緩沖區默認大小是1024個字節。所以，最多只能寫入1024個字節。
觀察PipedInputStream.java的源碼，我們能了解的更透徹。
|
1
2
3
4
|
private static final int DEFAULT_PIPE_SIZE = 1024;public PipedInputStream() { initPipe(DEFAULT_PIPE_SIZE);} |
默認構造函數調用initPipe(DEFAULT_PIPE_SIZE)，它的源碼如下：
|
1
2
3
4
5
6
|
private void initPipe(int pipeSize) { if (pipeSize <= 0) { throw new IllegalArgumentException("Pipe Size <= 0"); } buffer = new byte[pipeSize];} |
從中，我們可以知道緩沖區buffer的默認大小就是1024個字節。
試驗二： 在“試驗一”的基礎上繼續修改Receiver.java
將
|
1
2
3
4
|
public void run(){ readMessageOnce() ; //readMessageContinued() ;} |
修改為
|
1
2
3
4
|
public void run(){ //readMessageOnce() ; readMessageContinued() ;} |
運行程序。運行結果為：
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
012345678901234567890123456789abcd
efghijklmnopqrstuvwxyz
這個結果才是writeLongMessage()寫入到“輸入緩沖區”的完整數據。
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
