第九节 netty前传-NIO pipe

pipe管道作为线程之间通信的一种方式

• 首先作为对比我们先了解下再BIO模式下的pipe的使用
  Pipe为运行在同一个JVM中的两个线程提供了通信的能力，注意是同一个jvm上，如果在不同的jvm就是不同的进程了，pipe就无能为力了。

1. Java IO中管道的创建

Java IO中的PipedOutputStream和PipedInputStream创建管道。PipedInputStream流应该和PipedOutputStream流相关联。一个线程通过PipedOutputStream写入的数据可以被另一个线程通过相关联的PipedInputStream读取出来。

2. Java IO 管道的使用

read()方法和write()方法，分别是写入数据和读取数据，但是需要注意的是
这两个方法调用时会导致流阻塞，就是说如果尝试在一个线程中同时进行读和写，可能会导致线程死锁。

3. 管道输出流和管道输入流之间建立连接

PipedOutputStream outputPipe = new PipedOutputStream();
//构造器方式
PipedInputStream inputPipe = new PipedInputStream(output);
或者使用两者共有的方法connect()方法连接

4 完整示例如下：

public static void main(String[] args) throws IOException {
//分别创建输入和输出的管道流
        final PipedOutputStream outputPipe = new PipedOutputStream();
//PipedInputStream(output);构造器将管道数据流和管道输入流建立通信
        final PipedInputStream  inputPipe  = new PipedInputStream(output);
//在第一个线程将数据写入管道输出流
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    outputPipe.write("Hello world, pipe!".getBytes());
                } catch (IOException e) {
                }
            }
        });
//第二个线程从管道输入流中读取数据
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    int data = input.read();
                    while(data != -1){
                        System.out.print((char) data);
                        data = inputPipe .read();
                    }
                } catch (IOException e) {
                }
            }
        });
        thread1.start();
        thread2.start();
    }

---

• 下面为NIO模式下的管道使用
  管道pipe和线程之间的关系如下图：

  
  图片.png

由此可以看出NIO的管道和BIO模式下的管道是不同的，在NIO模式下没有输入输出流的概念但是使用发送sink和读取source的channe。使用同一个pipe实现线程之间数据的流转

1. 创建pipe管道

//多个线程之间使用同一个管道
Pipe pipe = Pipe.open();

2. 向管道写数据。先获取sinkChannel ，然后将数据写入sinkChannel

//1、先获取sinkChannel 
Pipe.SinkChannel sinkChannel = pipe.sink();
//2、将数据写入sinkChannel 
String newData = "New String to write to file..." + System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while(buf.hasRemaining()) {
    sinkChannel.write(buf);
}

3. 从管道中读取数据。先获取sourceChannel ，然后从sourceChannel 读取数据

//1、先获取sourceChannel 
Pipe.SourceChannel sourceChannel = pipe.source();
//2、然后从sourceChannel 读取数据
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = inChannel.read(buf);

2、3两步的读写分别在不同的线程实现

4. 完整实例：

public static void pipeExample(){
        Pipe pipe = null;
        ExecutorService exec = Executors.newFixedThreadPool(2);
        try{
//创建pipe
            pipe = Pipe.open();
            final Pipe pipeTemp = pipe;
//线程一向管道中写入数据
            exec.submit(new Callable<Object>(){
                @Override
                public Object call() throws Exception
                {
                    Pipe.SinkChannel sinkChannel = pipeTemp.sink();//向通道中写数据
                    while(true){
                        TimeUnit.SECONDS.sleep(1);
                        String newData = "hello world"+System.currentTimeMillis();
                        ByteBuffer buf = ByteBuffer.allocate(1024);
                        buf.clear();
                        buf.put(newData.getBytes());
                    //反转后可读
                        buf.flip();
                        while(buf.hasRemaining()){
                            System.out.println(buf);
                            sinkChannel.write(buf);
                        }
                    }
                }
            });
 //线程二读取管道中的数据
            exec.submit(new Callable<Object>(){
                @Override
                public Object call() throws Exception
                {
                    Pipe.SourceChannel sourceChannel = pipeTemp.source();//向通道中读数据
                    while(true){
                        TimeUnit.SECONDS.sleep(1);
                        ByteBuffer buf = ByteBuffer.allocate(1024);
                        buf.clear();
                    //可读大小
                        int bytesRead = sourceChannel.read(buf);
                        System.out.println("bytesRead="+bytesRead);
                        while(bytesRead >0 ){
                            buf.flip();
                            byte b[] = new byte[bytesRead];
                            int i=0;
                            while(buf.hasRemaining()){
                                b[i]=buf.get();
                                System.out.printf("%X",b[i]);
                                i++;
                            }
                            String s = new String(b);
                            System.out.println("========>>>>>>"+s);
                          //无数据时跳出当前循环体
                            bytesRead = sourceChannel.read(buf);
                        }
                    }
                }
            });
        }catch(IOException e){
            e.printStackTrace();
        }finally{
            exec.shutdown();
        }
    }