程式師世界 >> 編程語言 >> JAVA編程 >> 關於JAVA >> java io學習（三）管道的簡介,源碼分析和示例

java io學習（三）管道的簡介,源碼分析和示例

編輯：關於JAVA

管道(PipedOutputStream和PipedInputStream)的簡介,源碼分析和示例

本章，我們對java 管道進行學習。

java 管道介紹

在java中，PipedOutputStream和PipedInputStream分別是管道輸出流和管道輸入流。

它們的作用是讓多線程可以通過管道進行線程間的通訊。在使用管道通信時，必須將PipedOutputStream和PipedInputStream配套使用。

使用管道通信時，大致的流程是：我們在線程A中向PipedOutputStream中寫入數據，這些數據會自動的發送到與PipedOutputStream對應的PipedInputStream中，進而存儲在PipedInputStream的緩沖中；此時，線程B通過讀取PipedInputStream中的數據。就可以實現，線程A和線程B的通信。

PipedOutputStream和PipedInputStream源碼分析

下面介紹PipedOutputStream和PipedInputStream的源碼。在閱讀它們的源碼之前，建議先看看源碼後面的示例。待理解管道的作用和用法之後，再看源碼，可能更容易理解。

此外，由於在“java io系列03之 ByteArrayOutputStream的簡介,源碼分析和示例(包括OutputStream)”中已經對PipedOutputStream的父類OutputStream進行了介紹，這裡就不再介紹OutputStream。

在“java io系列02之 ByteArrayInputStream的簡介,源碼分析和示例(包括InputStream)”中已經對PipedInputStream的父類InputStream進行了介紹，這裡也不再介紹InputStream。

1. PipedOutputStream 源碼分析(基於jdk1.7.40)

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)

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

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

說明：

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

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();

}