标签：java css-selectors nonblocking pipe nio

我相信我做的一切正确.我创建一个管道,将接收器传递到编写器线程,使用OP_READ在选择器上注册源,启动选择器.一切正常,但是一旦我向接收器写入内容,就会出现管道异常的情况.为什么!!! ???这里没有破管.我烦了.我如何调试/了解这里发生了什么？有没有人有一个简单的管道示例,我可以运行它来测试是否正常.写在接收器上的线程,选择器读取它.

编辑：我几乎遵循建议here.很难在Internet中找到NIO管道的具体示例.

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;

public class SystemOutPipe extends Thread {

  public static void main(String[] args)
  {
    try {
      SystemOutPipe sop = new SystemOutPipe();
      sop.start();
      System.out.println("This message should be redirected to System.err\nNow waiting 5 seconds ...");
      Thread.sleep(5000L);
      sop.setStopped(true);
      sop.join();
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  private Selector selector;
  private Pipe pipe;
  private boolean stopped = false;

  public SystemOutPipe() throws IOException {
    super("SystemOutPipe");
    pipe = Pipe.open();
    System.setOut(new PrintStream(new PipeOutputStream(pipe)));
    selector = Selector.open();
    pipe.source().configureBlocking(false);
    pipe.source().register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
  }

  @Override
  public void run() {
    try {
      while (!isStopped()) {
        int n = selector.select(1L);
        if (n > 0) {
          Iterator<SelectionKey> it = selector.selectedKeys().iterator();
          while (it.hasNext()) {
            SelectionKey key = it.next();
            it.remove();
            if (key.isReadable()) {
              new ReadHandler(key).run();
            }
          }
        }
      }
    } catch (Exception e) {
      e.printStackTrace(); // writes to System.err !
    }
  }

  public synchronized boolean isStopped() {
    return stopped;
  }

  public synchronized void setStopped(final boolean stopped) {
    this.stopped = stopped;
  }

  public class ReadHandler implements Runnable {
    private final SelectionKey key;

    public ReadHandler(final SelectionKey key) {
      this.key = key;
    }

    @Override
    public void run() {
      ByteBuffer bbuf = (ByteBuffer) key.attachment();
      ReadableByteChannel channel = (ReadableByteChannel) key.channel();
      try
      {
        int count = 0;
        do {
          bbuf.clear();
          count = channel.read(bbuf);
          if (count > 0) System.err.write(bbuf.array(), 0, count);
        } while(count > 0);
      } catch (IOException e) {
        e.printStackTrace();
        key.cancel();
      }
    }
  }

  public class PipeOutputStream extends OutputStream {
    private final Pipe pipe;

    public PipeOutputStream(final Pipe pipe) {
      this.pipe = pipe;
    }

    @Override
    public void write(final int b) throws IOException {
      write(new byte[] { (byte) b });
    }

    @Override
    public void write(final byte[] b) throws IOException {
      write(b, 0, b.length);
    }

    @Override
    public void write(final byte[] b, final int off, final int len) throws IOException {
      ByteBuffer bbuf = ByteBuffer.wrap(b, off, len);
      bbuf.position(len);
      bbuf.flip();
      int count = 0;
      while (count < len) {
        int n = pipe.sink().write(bbuf);
        if (n == 0) {
          // let's wait a bit and not consume cpu
          try {
            Thread.sleep(1L);
          } catch (InterruptedException e) {
            throw new IOException(e);
          }
        }
        else count += n;
      }
    }
  }
}

例外：

java.io.IOException: Broken pipe
    at sun.nio.ch.FileDispatcher.write0(Native Method)
    at sun.nio.ch.FileDispatcher.write(FileDispatcher.java:39)
    at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:72)
    at sun.nio.ch.IOUtil.write(IOUtil.java:43)
    at sun.nio.ch.SinkChannelImpl.write(SinkChannelImpl.java:149)
    at com.niostuff.util.GCLogInterceptor.fileModified(GCLogInterceptor.java:180)
    at net.contentobjects.jnotify.linux.JNotifyAdapterLinux$WatchData.notifyFileModified(Unknown Source)
    at net.contentobjects.jnotify.linux.JNotifyAdapterLinux.notifyChangeEvent(Unknown Source)
    at net.contentobjects.jnotify.linux.JNotifyAdapterLinux$1.notify(Unknown Source)
    at net.contentobjects.jnotify.linux.JNotify_linux.callbackProcessEvent(Unknown Source)
    at net.contentobjects.jnotify.linux.JNotify_linux.nativeNotifyLoop(Native Method)
    at net.contentobjects.jnotify.linux.JNotify_linux.access$000(Unknown Source)
    at net.contentobjects.jnotify.linux.JNotify_linux$1.run(Unknown Source)

解决方法:

好的,所以我发现了问题.首先,感谢所有尝试提供帮助的人.希望你能从我的错误中学到东西.事件链是：

1-我没有耗尽接收缓冲区(源通道读取到的缓冲区),最终缓冲区已满.

2-现在它已满,pipeSourceChannel.read(readBuffer)返回0字节.有要读取的数据,但无法在完整的缓冲区上读取.

3-这导致通道被关闭(我自己在bytesRead == 0上这样做)和BrokenPipe.

我在这里学到的一课：管道很棘手.我认为非阻塞并发队列的使用要简单得多,就像这里提到的这个人一样：Java NIO Pipe vs BlockingQueue

标签：java,css-selectors,nonblocking,pipe,nio
来源： https://codeday.me/bug/20191010/1883826.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。