1

我正在尝试对一个程序进行原型设计，该程序将读取一个流，然后在线程池中启动新的处理流，因为主线程在流中走得更远。我遇到了 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>
4

1 回答 1

1

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

 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");

希望这可以帮助！

于 2013-05-05T20:00:01.940 回答