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 に答える