为什么我的 PipedOutputStream 死锁了？

Question

我正在尝试使用 PipedInputStream 和 PipedOutputStream 实现一个线程化的循环缓冲区，但每次我在 Decoder 可运行文件中到达 mHead.write 时它都会锁定。我认为使用单独的线程时不会出现死锁。

    private class DecoderTask implements Runnable{

    @Override
    public void run() {
        while(!mStop){
            try {
                    Log.d(TAG,"trying to write");
        mHead.write(decode( 0, 1000));
            mHead.flush();
            Log.d(TAG,"Decoded");
            } catch (DecoderException e) {
                Log.e(TAG,e.toString());
            } catch (IOException e) {
                Log.e(TAG,e.toString());
            }
        }
    }

}
private class WriteTask implements Runnable{

    @Override
    public void run() {
        while(!mStop){
            try {
                                 Log.d(TAG,"trying to read");
                 int read = mTail.read(mByteSlave, 0, mByteSlave.length);
                 mAudioTrack.flush();
                                 mAudioTrack.write(mByteSlave,0,read);
                                 Log.d(TAG,"read");                 
            } catch (IOException e) {
                Log.e(TAG,e.toString());
            }
        }
    }

}


//in some function
mTail = new PipedInputStream();
mHead = new PipedOutputStream(mTail);
mByteSlave = new byte[BUF];
mT1 = new Thread(new DecoderTask(), "Reader");
mT2 = new Thread(new WriteTask(), "Writer");
mT1.start();
mT2.start();
return;

编辑：这里是我的服务http://pastie.org/1179792的完整源代码

logcat 打印出来：

尝试阅读
试着写

Answer 1

我遇到了同样的问题，并通过覆盖PipedInputStream(int) 构造函数中的默认PIPE_SIZE 解决了它。方法PipedOutputStream.write(byte[], int, int) 阻塞，直到所有字节都写入输出流。这可能是默认 PIPE_SIZE 的问题。

毕竟，大小很重要 ;-)

Answer 2

该程序不会阻塞，它只是非常非常缓慢且效率低下。它使用 100% 的 CPU。问题是if (mTail.available() >= mByteSlave.length) - 在大多数情况下这将返回 false，因此您会在此线程中得到一个繁忙的循环。如果你能摆脱这个，那就去做吧。那么这个问题就解决了。如果你不能，它会变得更复杂......

还有一个问题：PipedInputStream.read 返回一个 int。你需要使用：

int len = mTail.read(mByteSlave, 0, mByteSlave.length);
mAudioTrack.write(mByteSlave, 0, len);

除此之外，我在这段代码中找不到任何错误。我的完整测试用例如下所示：

import java.io.*;
public class Test2 {
    PipedOutputStream mHead;
    PipedInputStream mTail;
    byte[] mByteSlave = new byte[1024];
    boolean mStop;
    public static void main(String... ar) throws Exception {
        new Test2().run();
    }
    void run() throws Exception {
        mTail = new PipedInputStream();
        mHead = new PipedOutputStream(mTail);
        Thread mT1 = new Thread(new DecoderTask(), "Reader");
        Thread mT2 = new Thread(new WriteTask(), "Writer");
        mT1.start();
        mT2.start();
    }
    class DecoderTask implements Runnable {
        public void run() {
            while (!mStop) {
                try {
                    mHead.write(new byte[3000]);
                    mHead.flush();
                    System.out.println("decoded 3000");
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    class WriteTask implements Runnable {
        public void run() {
            while (!mStop) {
                try {
                    int len = mTail.read(mByteSlave, 0, mByteSlave.length);
                    if (len < 0) break; // EOF
                    // mAudioTrack.write(mByteSlave, 0, len);
                    // mAudioTrack.flush();
                    System.out.println("written " + len);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

Answer 3

只需摆脱涉及 available() 的测试。无论如何，读取都会阻塞，当没有数据时你无所事事。