java - 管道流和倒计时锁存器

Question

我正在尝试构建一个程序原型(prototype)，该程序将读取流，然后随着主线程在流中进一步前进，在线程池中启动新的处理流。我遇到了 PipedStreams 和 CountDownLatch 的问题。

当我运行以下代码并在主线程中注释掉“latch.await()”时，出现“Write end dead”错误。当我使用未注释的“latch.await()”运行时，程序挂起并且没有线程退出。

知道我做错了什么或建议如何更好地使用多个处理流处理单个流。谢谢。

import java.io.BufferedInputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.concurrent.CountDownLatch;
//import java.util.Arrays;

public class RunTestPipePool {

    final static int BUFFER_SIZE = 8;

    static class PipeWriter extends Thread implements Runnable {
        InputStream in;
        OutputStream out;
        CountDownLatch latch;

        public PipeWriter(InputStream in, OutputStream out, CountDownLatch latch) {
            this.in = in;
            this.out = out;
            this.latch = latch;
        }

        public void run() {
            try {
                byte[] buffer = new byte[BUFFER_SIZE];
                int n = 0;
                while ((n = in.read(buffer)) >= 0) {
//                  System.out.println("PipeWriter Processing: " + new String(Arrays.copyOfRange(buffer,0,n)));
                    out.write(buffer,0,n);
                }
                latch.countDown();
                out.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            } finally {
                System.out.println("PipeWriter Terminating");
            }
        }
    }

    public static void main(String[] args) throws Exception {
        CountDownLatch latch = new CountDownLatch(1);

        PipedInputStream writeIn = new PipedInputStream();
        PipedOutputStream readOut = new PipedOutputStream(writeIn);

        InputStream reader = new BufferedInputStream(System.in);
        PipeWriter writer = new PipeWriter(writeIn,System.out,latch);

        writer.start();

        byte[] buffer = new byte[BUFFER_SIZE];
        int n = 0;
        while ((n = reader.read(buffer)) >= 0) {
//          System.out.println("RunTestPipePool Processing: " + new String(Arrays.copyOfRange(buffer,0,n)));
            readOut.write(buffer,0,n);
        }

        latch.await();
        reader.close();

        System.out.println("RunTestPipePool Terminating");
    }
}

带有latch.await()的输出被注释掉:

C:\Users\matty\Documents\workspace\test_pipe\bin>echo hello world | java -jar Ru
nTestPipePool.jar
RunTestPipePool Terminating
hello world
java.io.IOException: Write end dead
        at java.io.PipedInputStream.read(PipedInputStream.java:294)
        at java.io.PipedInputStream.read(PipedInputStream.java:361)
        at java.io.InputStream.read(InputStream.java:82)
        at RunTestPipePool$PipeWriter.run(RunTestPipePool.java:28)
PipeWriter Terminating

latch.await() 未注释的输出:

C:\Users\matty\Documents\workspace\test_pipe\bin>echo hello world | java -jar Ru
nTestPipePool.jar
hello world

C:\Users\matty\Documents\workspace\test_pipe\bin>

修改后的代码:

import java.io.BufferedInputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.concurrent.CountDownLatch;
//import java.util.Arrays;

public class RunTestPipeStack {

    final static int BUFFER_SIZE = 8;

    static class PipeWriter extends Thread implements Runnable {
        InputStream in;
        OutputStream out;
        CountDownLatch latch;

        public PipeWriter(InputStream in, OutputStream out, CountDownLatch latch) {
            this.in = in;
            this.out = out;
            this.latch = latch;
        }

        public void run() {
            try {
                byte[] buffer = new byte[BUFFER_SIZE];
                int n = 0;
                while (in.available() != 0 && (n = in.read(buffer)) >= 0) {
//                  System.out.println("PipeWriter Processing: " + new String(Arrays.copyOfRange(buffer,0,n)));
                    out.write(buffer,0,n);
                }
                System.out.println("PipeWriter Terminating");
                in.close();
                out.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        CountDownLatch latch = new CountDownLatch(1);

        PipedInputStream writeIn = new PipedInputStream();
        PipedOutputStream readOut = new PipedOutputStream(writeIn);

        InputStream reader = new BufferedInputStream(System.in);
        PipeWriter writer = new PipeWriter(writeIn,System.out,latch);

        writer.start();

        byte[] buffer = new byte[BUFFER_SIZE];
        int n = 0;
        while ((n = reader.read(buffer)) >= 0) {
            //          System.out.println("RunTestPipePool Processing: " + new String(Arrays.copyOfRange(buffer,0,n)));
            readOut.write(buffer,0,n);
        }

        reader.close();

        System.out.println("RunTestPipePool Terminating");
    }
}

修改后的输出:

C:\Users\matty\Documents\workspace\test_pipe\bin>echo hello world | java -jar Ru
nTestPipeStack.jar
RunTestPipePool Terminating
hello world
PipeWriter Terminating

C:\Users\matty\Documents\workspace\test_pipe\bin>

Answer 1

await() 的使用是正确的，但它不起作用的原因是线程处于 in.read() 的阻塞状态 - 等待下一组字节。但在“hello world”之后，就没有任何内容了。 完成读取后，检查是否有更多可用字节。如果它们没有关闭流并跳出循环。另外，您可能想添加 in!=null 检查 while 循环。

 while (in!=null && (n = in.read(buffer)) >= 0) {
               System.out.println("PipeWriter Processing: " + new  
                                String(Arrays.copyOfRange(buffer,0,n)));
                out.write(buffer,0,n);


                if(in.available()==0)   
                {
                    latch.countDown();
                    out.close();    
                    in.close();
                    break;
                }

            }
            System.out.println("Completed the PipeWriter loop");

希望这有帮助!