管道Pipe

管道Pipe

　　java.nio.channels包中含有一個名爲Pipe（管道）的類。廣義上講，管道就是一個用來在兩個實體之間單向傳輸數據的導管。管道的概念對於Unix（和類Unix）操做系統的用戶來講早就很熟悉了。Unix系統中，管道被用來鏈接一個進程的輸出和另外一個進程的輸入。Pipe類實現一個管道範例，不過它所建立的管道是進程內（在Java虛擬機進程內部）而非進程間使用的。

參見圖3-10。

 

/*
 * @(#)Pipe.java    1.21 05/11/17
 *
 * Copyright 2006 Sun Microsystems, Inc. All rights reserved.
 * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */

package java.nio.channels;

import java.io.IOException;
import java.nio.channels.spi.*;


/**
 * A pair of channels that implements a unidirectional pipe.
 *
 * <p> A pipe consists of a pair of channels: A writable {@link
 * Pipe.SinkChannel </code>sink<code>} channel and a readable {@link
 * Pipe.SourceChannel </code>source<code>} channel.  Once some bytes are
 * written to the sink channel they can be read from source channel in exactly
 * the order in which they were written.
 *
 * <p> Whether or not a thread writing bytes to a pipe will block until another
 * thread reads those bytes, or some previously-written bytes, from the pipe is
 * system-dependent and therefore unspecified.  Many pipe implementations will
 * buffer up to a certain number of bytes between the sink and source channels,
 * but such buffering should not be assumed.  </p>
 *
 *
 * @author Mark Reinhold
 * @author JSR-51 Expert Group
 * @version 1.21, 05/11/17
 * @since 1.4
 */

public abstract class Pipe {

    /**
     * A channel representing the readable end of a {@link Pipe}.  </p>
     *
     * @since 1.4
     */
    public static abstract class SourceChannel
    extends AbstractSelectableChannel
    implements ReadableByteChannel, ScatteringByteChannel
    {
    /**
     * Constructs a new instance of this class.
     */
    protected SourceChannel(SelectorProvider provider) {
        super(provider);
    }

    /**
     * Returns an operation set identifying this channel's supported
     * operations.
     *
     * <p> Pipe-source channels only support reading, so this method
     * returns {@link SelectionKey#OP_READ}.  </p>
     *
     * @return  The valid-operation set
     */
    public final int validOps() {
        return SelectionKey.OP_READ;
    }

    }

    /**
     * A channel representing the writable end of a {@link Pipe}.  </p>
     *
     * @since 1.4
     */
    public static abstract class SinkChannel
    extends AbstractSelectableChannel
    implements WritableByteChannel, GatheringByteChannel
    {
    /**
     * Initializes a new instance of this class.
     */
    protected SinkChannel(SelectorProvider provider) {
        super(provider);
    }

    /**
     * Returns an operation set identifying this channel's supported
     * operations.
     *
     * <p> Pipe-sink channels only support writing, so this method returns
     * {@link SelectionKey#OP_WRITE}.  </p>
     *
     * @return  The valid-operation set
     */
    public final int validOps() {
        return SelectionKey.OP_WRITE;
    }

    }

    /**
     * Initializes a new instance of this class.
     */
    protected Pipe() { }

    /**
     * Returns this pipe's source channel.  </p>
     *
     * @return  This pipe's source channel
     */
    public abstract SourceChannel source();

    /**
     * Returns this pipe's sink channel.  </p>
     *
     * @return  This pipe's sink channel
     */
    public abstract SinkChannel sink();

    /**
     * Opens a pipe.
     *
     * <p> The new pipe is created by invoking the {@link
     * java.nio.channels.spi.SelectorProvider#openPipe openPipe} method of the
     * system-wide default {@link java.nio.channels.spi.SelectorProvider}
     * object.  </p>
     *
     * @return  A new pipe
     *
     * @throws  IOException
     *          If an I/O error occurs
     */
    public static Pipe open() throws IOException {
    return SelectorProvider.provider().openPipe();
    }

}

　　Pipe實例是經過調用不帶參數的Pipe.open( )工廠方法來建立的。Pipe類定義了兩個嵌套的通道類來實現管路。這兩個類是Pipe.SourceChannel（管道負責讀的一端）和Pipe.SinkChannel（管道負責寫的一端）。這兩個通道實例是在Pipe對象建立的同時被建立的，能夠經過在Pipe對象上分別調用source( )和sink( )方法來取回。

　　此時，您可能在想管道到底有什麼做用。您不能使用Pipe在操做系統級的進程間創建一個類Unix管道（您可使用SocketChannel來創建）。Pipe的source通道和sink通道提供相似java.io.PipedInputStream和java.io.PipedOutputStream所提供的功能，不過它們能夠執行所有的通道語義。請注意，SinkChannel和SourceChannel都繼承了AbstractSelectableChannel（因此也間接地繼承了SelectableChannel），這意味着pipe通道能夠同選擇器一塊兒使用 。

　　管道能夠被用來僅在同一個Java虛擬機內部傳輸數據。雖然有更加有效率的方式來在線程之間傳輸數據，可是使用管道的好處在於封裝性。生產者線程和用戶線程都能被寫道通用的Channel API中。根據給定的通道類型，相同的代碼能夠被用來寫數據到一個文件、socket或管道。選擇器能夠被用來檢查管道上的數據可用性，如同在socket通道上使用那樣地簡單。這樣就能夠容許單個用戶線程使用一個Selector來從多個通道有效地收集數據，並可任意結合網絡鏈接或本地工做線程使用。所以，這些對於可伸縮性、冗餘度以及可複用性來講無疑都是意義重大的。

　　Pipes的另外一個有用之處是能夠用來輔助測試。一個單元測試框架能夠將某個待測試的類鏈接到管道的「寫」端並檢查管道的「讀」端出來的數據。它也能夠將被測試的類置於通道的「讀」端並將受控的測試數據寫進其中。兩種場景對於迴歸測試都是頗有幫助的。

　　管路所能承載的數據量是依賴實現的（implementation-dependent）。惟一可保證的是寫到SinkChannel中的字節都能按照一樣的順序在SourceChannel上重現。

　　例3-11詮釋瞭如何使用管道。

/**
 * 
 */
package test.nio.pipe;

import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.Pipe;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.Random;

/**
 * * Test Pipe objects using a worker thread. ** Created April, 2002 * @author
 * Ron Hitchens (ron@ronsoft.com)
 */
public class PipeTest {
    public static void main(String[] argv) throws Exception {
        // Wrap a channel around stdout
        WritableByteChannel out = Channels.newChannel(System.out);
        // Start worker and get read end of channel
        ReadableByteChannel workerChannel = startWorker(10);
        ByteBuffer buffer = ByteBuffer.allocate(100);
        while (workerChannel.read(buffer) >= 0) {
            buffer.flip();
            out.write(buffer);
            buffer.clear();
        }
    }

    // This method could return a SocketChannel or
    // FileChannel instance just as easily
    private static ReadableByteChannel startWorker(int reps) throws Exception {
        Pipe pipe = Pipe.open();
        Worker worker = new Worker(pipe.sink(), reps);
        worker.start();
        return (pipe.source());
    }// -----------------------------------------------------------------

    /**
     * * A worker thread object which writes data down a channel. * Note: this
     * object knows nothing about Pipe, uses only a * generic
     * WritableByteChannel.
     */
    private static class Worker extends Thread {
        WritableByteChannel channel;
        private int reps;

        Worker(WritableByteChannel channel, int reps) {
            this.channel = channel;
            this.reps = reps;
        }

        // Thread execution begins here
        public void run() {
            ByteBuffer buffer = ByteBuffer.allocate(100);
            try {
                for (int i = 0; i < this.reps; i++) {
                    doSomeWork(buffer);
                    // channel may not take it all at once
                    while (channel.write(buffer) > 0) {
                        // empty
                    }
                }
                this.channel.close();
            } catch (Exception e) {
                // easy way out; this is demo code
                e.printStackTrace();
            }
        }

        private String[] products = { "No good deed goes unpunished",
                "To be, or what?", "No matter where you go, there you are",
                "Just say \"Yo\"", "My karma ran over my dogma" };
        private Random rand = new Random();

        private void doSomeWork(ByteBuffer buffer) {
            int product = rand.nextInt(products.length);
            buffer.clear();
            buffer.put(products[product].getBytes());
            buffer.put("\r\n".getBytes());
            buffer.flip();
        }
    }
}

　　例3-11 工做線程對一個管道進行寫操做

 

 

以上內容出自 : NIO 一書