程式師世界 >> 編程語言 >> JAVA編程 >> 關於JAVA >> 舉例講授Java中Piped管道輸出輸入流的線程通訊掌握

舉例講授Java中Piped管道輸出輸入流的線程通訊掌握

編輯：關於JAVA

舉例講授Java中Piped管道輸出輸入流的線程通訊掌握。本站提示廣大學習愛好者：（舉例講授Java中Piped管道輸出輸入流的線程通訊掌握）文章只能為提供參考，不一定能成為您想要的結果。以下是舉例講授Java中Piped管道輸出輸入流的線程通訊掌握正文

PipedOutputStream和PipedInputStream

在java中，PipedOutputStream和PipedInputStream分離是管道輸入流和管道輸出流。
它們的感化是讓多線程可以經由過程管道停止線程間的通信。在應用管道通訊時，必需將PipedOutputStream和PipedInputStream配套應用。
應用管道通訊時，年夜致的流程是：我們在線程A中向PipedOutputStream中寫入數據，這些數據會主動的發送到與PipedOutputStream對應的PipedInputStream中，進而存儲在PipedInputStream的緩沖中；此時，線程B經由過程讀取PipedInputStream中的數據。便可以完成，線程A和線程B的通訊。
上面，我們看看多線程中經由過程管道通訊的例子。例子中包含3個類：Receiver.java, PipedStreamTest.java 和 Sender.java。
Receiver.java的代碼以下：

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的代碼以下：

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的代碼以下：

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

解釋：
(1) in.connect(out);將“管道輸出流”和“管道輸入流”聯系關系起來。檢查PipedOutputStream.java和PipedInputStream.java中connect()的源碼；我們曉得 out.connect(in); 等價於 in.connect(out);
(2)

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[])的源碼以下：

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
將

public void run(){ 
 writeShortMessage();
 //writeLongMessage();
}

修正為

public void run(){ 
 //writeShortMessage();
 writeLongMessage();
}

運轉法式。運轉成果為：

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
012345678901234567890123456789abcd

這些數據是經由過程writeLongMessage()寫入到“管道輸入流”，然後傳送給“管道輸出流”，進而存儲在“管道輸出流”的緩沖中；再被用戶從緩沖讀掏出來的數據。
然後，不雅察writeLongMessage()的源碼。我們可以發明，str的長度是1046個字節，然後運轉成果只要1024個字節！為何會如許呢？
事理很簡略：管道輸出流的緩沖區默許年夜小是1024個字節。所以，最多只能寫入1024個字節。
不雅察PipedInputStream.java的源碼，我們能懂得的更透辟。

private static final int DEFAULT_PIPE_SIZE = 1024;
public PipedInputStream() {
 initPipe(DEFAULT_PIPE_SIZE);
}

默許結構函數挪用initPipe(DEFAULT_PIPE_SIZE)，它的源碼以下：

private void initPipe(int pipeSize) {
  if (pipeSize <= 0) {
  throw new IllegalArgumentException("Pipe Size <= 0");
  }
  buffer = new byte[pipeSize];
}

從中，我們可以曉得緩沖區buffer的默許年夜小就是1024個字節。
實驗二：在“實驗一”的基本上持續修正Receiver.java
將

public void run(){ 
 readMessageOnce() ;
 //readMessageContinued() ;
}

修正為

public void run(){ 
 //readMessageOnce() ;
 readMessageContinued() ;
}

運轉法式。運轉成果為：

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
012345678901234567890123456789abcd
efghijklmnopqrstuvwxyz

這個成果才是writeLongMessage()寫入到“輸出緩沖區”的完全數據。

PipedWriter和PipedReader
PipedWriter 是字符管道輸入流，它繼續於Writer。
PipedReader 是字符管道輸出流，它繼續於Writer。
PipedWriter和PipedReader的感化是可以經由過程管道停止線程間的通信。在應用管道通訊時，必需將PipedWriter和PipedReader配套應用。
上面，我們看看多線程中經由過程PipedWriter和PipedReader通訊的例子。例子中包含3個類：Receiver.java, Sender.java 和 PipeTest.java
Receiver.java的代碼以下：

import java.io.IOException; 

import java.io.PipedReader; 

@SuppressWarnings("all") 
/** 
 * 吸收者線程 
 */ 
public class Receiver extends Thread { 

 // 管道輸出流對象。
 // 它和“管道輸入流(PipedWriter)”對象綁定，
 // 從而可以吸收“管道輸入流”的數據，再讓用戶讀取。
 private PipedReader in = new PipedReader(); 

 // 取得“管道輸出流對象”
 public PipedReader getReader(){ 
  return in; 
 } 

 @Override
 public void run(){ 
  readMessageOnce() ;
  //readMessageContinued() ;
 }

 // 從“管道輸出流”中讀取1次數據
 public void readMessageOnce(){ 
  // 固然buf的年夜小是2048個字符，但最多只會從“管道輸出流”中讀取1024個字符。
  // 由於，“管道輸出流”的緩沖區年夜小默許只要1024個字符。
  char[] buf = new char[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) {
   char[] buf = new char[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的代碼以下：

import java.io.IOException; 

import java.io.PipedWriter; 
@SuppressWarnings("all")
/** 
 * 發送者線程 
 */ 
public class Sender extends Thread { 

 // 管道輸入流對象。
 // 它和“管道輸出流(PipedReader)”對象綁定，
 // 從而可以將數據發送給“管道輸出流”的數據，然後用戶可以從“管道輸出流”讀取數據。
 private PipedWriter out = new PipedWriter();

 // 取得“管道輸入流”對象
 public PipedWriter getWriter(){
  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.toCharArray());
   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);
   out.close();
  } catch (IOException e) {
   e.printStackTrace();
  }
 }
}

PipeTest.java的代碼以下：

import java.io.PipedReader;
import java.io.PipedWriter;
import java.io.IOException;

@SuppressWarnings("all") 
/** 
 * 管道輸出流和管道輸入流的交互法式
 */ 
public class PipeTest { 

 public static void main(String[] args) { 
  Sender t1 = new Sender(); 

  Receiver t2 = new Receiver(); 

  PipedWriter out = t1.getWriter(); 

  PipedReader in = t2.getReader(); 

  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
成果解釋：

(1)

 in.connect(out);

它的感化是將“管道輸出流”和“管道輸入流”聯系關系起來。檢查PipedWriter.java和PipedReader.java中connect()的源碼；我們曉得 out.connect(in); 等價於 in.connect(out);
(2)

t1.start(); // 啟動“Sender”線程
t2.start(); // 啟動“Receiver”線程

先檢查Sender.java的源碼，線程啟動後履行run()函數；在Sender.java的run()中，挪用writeShortMessage();
writeShortMessage();的感化就是向“管道輸入流”中寫入數據"this is a short message" ；這條數據會被“管道輸出流”吸收到。上面看看這是若何完成的。
先看write(char char的源碼。PipedWriter.java繼續於Writer.java；Writer.java中write(char c[])的源碼以下：

public void write(char cbuf[]) throws IOException {
 write(cbuf, 0, cbuf.length);
}

現實上write(char c[])是挪用的PipedWriter.java中的write(char c[], int off, int len)函數。檢查write(char c[], int off, int len)的源碼，我們發明：它會挪用 sink.receive(cbuf, off, len); 進一步檢查receive(char c[], int off, int len)的界說，我們曉得sink.receive(cbuf, 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
將

public void run(){ 
 writeShortMessage();
 //writeLongMessage();
}

修正為

public void run(){ 
 //writeShortMessage();
 writeLongMessage();
}

運轉法式。運轉成果以下：

從中，我們看出，法式運轉失足！拋出異常 java.io.IOException: Pipe closed
為何會如許呢？
我剖析一下法式流程。
(1) 在PipeTest中，經由過程in.connect(out)將輸出和輸入管道銜接起來；然後，啟動兩個線程。t1.start()啟動了線程Sender，t2.start()啟動了線程Receiver。
(2) Sender線程啟動後，經由過程writeLongMessage()寫入數據到“輸入管道”，out.write(str.toCharArray())共寫入了1046個字符。而依據PipedWriter的源碼，PipedWriter的write()函數會挪用PipedReader的receive()函數。而不雅察PipedReader的receive()函數，我們曉得，PipedReader會將接收的數據存儲緩沖區。細心不雅察receive()函數，有以下代碼：

while (in == out) {
 if ((readSide != null) && !readSide.isAlive()) {
  throw new IOException("Pipe broken");
 }
 /* full: kick any waiting readers */
 notifyAll();
 try {
  wait(1000);
 } catch (InterruptedException ex) {
  throw new java.io.InterruptedIOException();
 }
}

而in和out的初始值分離是in=-1, out=0；聯合下面的while(in==out)。我們曉得，它的寄義就是，每往管道中寫入一個字符，就到達了in==out這個前提。然後，就挪用notifyAll()，叫醒“讀取管道的線程”。
也就是，每往管道中寫入一個字符，都邑壅塞式的期待其它線程讀取。
但是，PipedReader的緩沖區的默許年夜小是1024！然則，此時要寫入的數據卻有1046！所以，一次性最多只能寫入1024個字符。
(03) Receiver線程啟動後，會挪用readMessageOnce()讀取管道輸出流。讀取1024個字符會，會挪用close()封閉，管道。
由(02)和(03)的剖析可知，Sender要往管道寫入1046個字符。個中，前1024個字符(緩沖區容量是1024)能正常寫入，而且每寫入一個就讀取一個。當寫入1025個字符時，仍然是順次的挪用PipedWriter.java中的write()；然後，write()中挪用PipedReader.java中的receive()；在PipedReader.java中，終究又會挪用到receive(int c)函數。而此時，管道輸出流曾經被封閉，也就是closedByReader為true，所以拋出throw new IOException("Pipe closed")。
我們對“實驗一”持續停止修正，處理該成績。
實驗二：在“實驗一”的基本上持續修正Receiver.java 將

public void run(){ 
 readMessageOnce() ;
 //readMessageContinued() ;
}

修正為

public void run(){ 
 //readMessageOnce() ;
 readMessageContinued() ;
}

此時，法式能正常運轉。運轉成果為：

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789
012345678901234567890123456789abcd
efghijklmnopqrstuvwxyz