NIO系列（六）——线程通信(pipe)

清疚 2022-06-13 02:16 313阅读 0赞

## Pipe ##

Pipe用于线程之间的数据传输。Pipe有两个嵌套类(内部类)SinkChannel和SourceChannel，数据写入到SinkChannel，数据从SourceChannel读取。

### 创建Pipe ###

通过Pipe.open()来创建管道

Pipe pipe = Pipe.open();

### 向管道写入数据 ###

创建SinkChannel对象

Pipe.SinkChannel sink = pipe.sink();

调用SinkChannel的write()方法，向SinkChannel写入数据，就像这样

String writeData = "threadA";
    System.out.println("线程A输入数据：" + writeData);
    ByteBuffer byteBuffer =ByteBuffer.allocate(64);
    byteBuffer.put(writeData.getBytes("UTF-8"));
    byteBuffer.flip();
    while (byteBuffer.hasRemaining()) {
        sink.write(byteBuffer);
    }
    byteBuffer.clear();
    sink.close();

### 从管道读出数据 ###

创建SourceChannel对象

Pipe.SourceChannel source = pipe.source();

调用SourceChannel的read()方法读取数据，就像这样

ByteBuffer byteBuffer = ByteBuffer.allocate(64);
        int readSize = source.read(byteBuffer);
        byteBuffer.flip();
        while (readSize != -1) {
           System.out.println(Charset.forName("UTF-8").decode(byteBuffer));
           byteBuffer.clear();
           readSize = source.read(byteBuffer);
        }
    source.close();

### 例子 ###

这个简单的例子实现了线程A传数据到线程B的过程，运行main函数就可以。

public staticvoidmain(String[] args){
        try {
           Pipepipe= Pipe.open();
           Pipe.SinkChannelsink= pipe.sink();
           Pipe.SourceChannelsource= pipe.source();
           ThreadthreadA= newThread() {
               public void run() {
                  StringwriteData= "threadA";
                  System.out.println("线程A输入数据：" + writeData);
                  ByteBufferbyteBuffer= ByteBuffer.allocate(64);
                  try {
                      byteBuffer.put(writeData.getBytes("UTF-8"));
                      byteBuffer.flip();
                      while (byteBuffer.hasRemaining()) {
                         sink.write(byteBuffer);
                      }
                      byteBuffer.clear();
                      sink.close();
                  }catch(UnsupportedEncodingException e) {
                      e.printStackTrace();
                  }catch(IOException e) {
                      e.printStackTrace();
                  }
               }
           };
           ThreadthreadB= newThread() {
               public void run() {
                  ByteBufferbyteBuffer= ByteBuffer.allocate(64);
                  try {
                      int readSize = source.read(byteBuffer);
                      byteBuffer.flip();
                      while (readSize != -1) {
                         System.out.println(Charset.forName("UTF-8").decode(byteBuffer));
                         byteBuffer.clear();
                         readSize = source.read(byteBuffer);
                      }
                      source.close();
                  }catch(UnsupportedEncodingException e) {
                      e.printStackTrace();
                  }catch(IOException e) {
                      e.printStackTrace();
                  }
               }
           };
           threadA.start();
           threadB.start();
        }catch(IOException e) {
           e.printStackTrace();
        }
    }