java - 在 InputStream 的 read() 中阻止 I/O

Question

我正在尝试编写一种算法，用于下载视频直播流。具体来说，我尝试获取的相应流基于动态 .m3u8 播放列表文件，该文件定期提供新视频文件的 URI。主要目标是将这些单独的媒体文件组合成一个连贯的输入流。
我实际上成功地让它工作:我定期检查播放列表中出现的新媒体文件，并将它们的 HTTP 流传递给自定义的 InputStream 实现，即 InputStreamChain 。由于这是一场直播，我认为它会无休无止，至少目前是这样。因此，我希望我的 InputStreamChain 的 read() 永远不会发送 -1。不幸的是，它确实发生了；每次当所有排队的媒体流都被消耗完时，InputStreamChain 就会结束。相反，我希望它阻止 I/O，直到新的媒体文件到达。因此，我想出了一个可行的解决方案:我调整了 read() 方法以进行循环，直到有新的流可用(TimerTask 将提供新文件)。在循环中，我内置了一个Thread.sleep()，以减少CPU负载:

public int read() throws IOException {  
    int bit = current.read();  
    if (bit == -1 && streams.size() > 0) {  
        // left out due to lacking relevance
    } else if(bit == -1 && streams.size() == 0) {
        while(streams.size() == 0) {
            Thread.currentThread().sleep(50);
        }
        return read();
    }
    return bit;
}

虽然这似乎有效，但我有一种感觉，我没有按照我应该做的方式去做。我还尝试将 Lock 与 Condition.await() 一起使用，但是当我的 TimerTask 尝试触发 Condition.signal()，它只是抛出了一个 IllegalMonitorStateException。
这就是为什么我要问这个问题:

我应该以什么方式延迟/阻止 InputStream 的 read() 方法，特别是在我的场景中？

编辑:

为了完整起见，我还将提供失败的 Lock 方法:

private ReentrantLock ioLock;
private Condition ioCond;
private boolean waitingForStream = false;

public InputStreamChain() {
    ioLock = new ReentrantLock();
    ioCond = ioLock.newCondition();
}

public synchronized InputStreamChain addInputStream(final InputStream stream) {
    streams.addLast(stream);
    if (current == null) {
        current = streams.removeFirst();
    }
    if(waitingForStream) {
        ioCond.signal();
    }
    return this;
}

public int read() throws IOException {
    int bit = current.read();
    if (bit == -1 && streams.size() > 0) {
        // do stuff
    } else if(bit == -1) {
        waitingForStream = true;
        ioLock.lock();
        try {
            ioCond.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            waitingForStream = false;
            ioLock.unlock();
        }
        return read();
    }
    return bit;
}

Answer 1

可能您没有使用同步块(synchronized block)。这是一个例子:

class MyReader
{
     public int read() throws IOException {  
        int bit = current.read();  
        if (bit == -1 && streams.size() > 0) {  
            // left out due to lacking relevance
        } else if(bit == -1 && streams.size() == 0) {
            waitForNextStream();
            return read();
        }
        return bit;
    }

    private synchronized void waitForNextStream()
    {
        // TODO add close handling, set current here
        while (streams.isEmpty())
        {
            wait();
        }
    }


    public synchronized void addNextStream(InputStream is)
    {
        streams.add(is);
        notify();
    }
}