TIJ -- 任务间使用管道进行输入/输出

Posted limeOracle

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了TIJ -- 任务间使用管道进行输入/输出相关的知识,希望对你有一定的参考价值。

  1. 通过输入/输出在线程间进行通信通常很有用。提供线程功能的类库以“管道”的形式对线程间的输入/输出提供了支持。它们在Java输入/输出类库中的对应物就是PipedWriter类(允许任务向管道写)和PipedReader类(允许不同任务从同一个管道中读取)。这个模型可以看成是“生产者 - 消费者”问题的变体,这里的管道就是一个封装好的解决方案。管道基本上是一个阻塞队列,存在于多个引入BlockingQueue之前的Java版本中。

  2. 下面是一个简单例子,两个任务使用一个管道进行通信:

    Class : 

package lime.thinkingInJava._021._005._005;

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;

/**
 * @Author : Lime
 * @Description :
 * @Remark :
 */
class Sender implements Runnable{
    private Random rand = new Random(47);
    private PipedWriter out = new PipedWriter();
    public PipedWriter getPipedWriter(){
        return out;
    }
    public void run(){
        try{
            while (true){
                for(char c = ‘A‘; c <= ‘z‘;c++){
                    out.write(c);
                    TimeUnit.MILLISECONDS.sleep(rand.nextInt(500));
                }
            }
        } catch (InterruptedException e) {
            System.out.println(e + " Sender sleep Interrupted");
        } catch (IOException e) {
            System.out.println(e + " Sender write exception");
        }
    }
}
class Receiver implements Runnable{
    private PipedReader in;
    public Receiver(Sender sender) throws IOException {
        in = new PipedReader(sender.getPipedWriter());
    }
    public void run(){
        try{
            while (true){
                //Blocks until characters are there;
                System.out.println("Read : " + (char)in.read());
            }
        } catch (IOException e) {
            System.out.println(e + " Receiver read exception");
        }
    }
}
public class PipedIO {
    public static void main(String[] args) throws IOException, InterruptedException {
        Sender sender = new Sender();
        Receiver receiver = new Receiver(sender);
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(sender);
        exec.execute(receiver);
        TimeUnit.SECONDS.sleep(4);
        exec.shutdownNow();
    }
}

  3. Console : 

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
java.lang.InterruptedException: sleep interrupted Sender sleep Interrupted
java.io.InterruptedIOException Receiver read exception

  4. Sender和Receiver代表了需要互相通信两个任务。Sender创建了一个PipedWriter,它是一个单独的对象;但是对于Receiver,PipedReader的建立必须在构造器中与一个PipedWriter相关联。Sender把数据放进Writer,然后休眠一段时间(随机数)。然而,Receiver没有Sleep()和wait()。但当它调用read()时,如果没有更多的数据,管道将自动阻塞。

  注意sender和receiver是在main()中启动的,即对象构造彻底完毕以后。如果你启动了一个没有构造完毕的对象,在不同的平台上管道可能会产生不一致的行为(注意,BlockingQueue使用起来更加健壮而容易)。

  在shutdownNow()被调用时,可以看到PipedReader与普通I/O之间最重要的差异 ------ PipedReader是可中断的。如果你将in.read()调用修改为System.in.read(),那么interrupt()将不能打断read()调用。

  5. PipedWriter的wirte() 源码解析

    /**
     * Writes the specified <code>char</code> to the piped output stream.
     * If a thread was reading data characters from the connected piped input
     * stream, but the thread is no longer alive, then an
     * <code>IOException</code> is thrown.
     * <p>
     * Implements the <code>write</code> method of <code>Writer</code>.
     *
     * @param      c   the <code>char</code> to be written.
     * @exception  IOException  if the pipe is
     *          <a href=PipedOutputStream.html#BROKEN> <code>broken</code></a>,
     *          {@link #connect(java.io.PipedReader) unconnected}, closed
     *          or an I/O error occurs.
     */
    public void write(int c)  throws IOException {
        if (sink == null) {
            throw new IOException("Pipe not connected");
        }
        //调用PipedReader的receive(c)方法,将c放入PipedReader的char buffer[]中
        sink.receive(c);
    }
    /**
     * Receives a char of data. This method will block if no input is
     * available.
     */
    synchronized void receive(int c) throws IOException {
        if (!connected) {
            //判断两个I/O流连接状态
            throw new IOException("Pipe not connected");
        } else if (closedByWriter || closedByReader) {
            //判断两个I/O流开启状态
            throw new IOException("Pipe closed");
        } else if (readSide != null && !readSide.isAlive()) {
            //判断输入流线程是否存活
            throw new IOException("Read end dead");
        }

        //获取输出流线程
        writeSide = Thread.currentThread();
        while (in == out) {
            //判断char buffer[] 是否已满
            if ((readSide != null) && !readSide.isAlive()) {
                //判断输入流状态是否存活
                throw new IOException("Pipe broken");
            }
            /* full: kick any waiting readers */
            // 间隔1000毫秒唤醒写线程 -- start
            notifyAll();
            try {
                //阻塞1000毫秒
                wait(1000);
            } catch (InterruptedException ex) {
                throw new java.io.InterruptedIOException();
            }
            // 间隔1000毫秒唤醒写线程 -- end
        }
        if (in < 0) {
            //判断char buffer[] 为空
            in = 0;
            out = 0;
        }
        buffer[in++] = (char) c;
        if (in >= buffer.length) {
            in = 0;
        }
    }

  6. 

 

 

 

 

 

啦啦啦

以上是关于TIJ -- 任务间使用管道进行输入/输出的主要内容,如果未能解决你的问题,请参考以下文章

Java多线程-管道流实现线程间通信

JAVA的IO编程:管道流

Linux进程间通信-命名管道

线程间通讯-管道输入输出流(源码分析)

C中的进程间通信

使用 FFmpeg 通过管道输出视频片段