如何用Okio Pipe实现固定大小循环缓冲Sink？解决写入阻塞问题

嘿，我来帮你捋清楚这个问题～

首先，Okio的Pipe本质是个阻塞式的同步通道，它的设计逻辑就是生产者写数据，必须等消费者读走一部分后，生产者才能继续写——当Pipe内部缓冲区满了，没有消费者读取的话，write()调用就会直接卡住，这是它的预期行为，不是bug。

而你要的是“固定大小的滑动窗口缓冲”（满了就删掉最早的旧数据，塞新数据进去），这种自动覆盖的逻辑，Pipe自带的BufferedSink根本做不到，因为它从设计上就没有“覆盖旧数据”的机制。

那该怎么实现你要的预缓冲逻辑呢？这里给你两种可行的思路：

方式一：自定义滑动窗口Sink（配合Pipe） #

我们可以自己封装一个带滑动缓冲逻辑的Sink，先把音频数据处理成“只保留最新固定大小”的状态，再写入Pipe的Sink。这样即使Pipe暂时满了，只要消费者持续读取数据，生产者就不会被阻塞。

import okio.Buffer;
import okio.Sink;
import okio.Timeout;
import java.io.IOException;

public class SlidingWindowSink implements Sink {
    private final Sink delegateSink; // 实际对接的Pipe Sink
    private final long maxBufferSize;
    private final Buffer slidingBuffer = new Buffer();
    private final Object lock = new Object(); // 线程安全锁，毕竟音频数据一般来自后台线程

    public SlidingWindowSink(Sink delegateSink, long maxBufferSize) {
        this.delegateSink = delegateSink;
        this.maxBufferSize = maxBufferSize;
    }

    @Override
    public void write(Buffer source, long byteCount) throws IOException {
        synchronized (lock) {
            // 先把新数据塞进滑动缓冲
            slidingBuffer.write(source, byteCount);

            // 如果缓冲超了最大容量，删掉最早的多余数据
            if (slidingBuffer.size() > maxBufferSize) {
                long excessBytes = slidingBuffer.size() - maxBufferSize;
                slidingBuffer.skip(excessBytes);
            }

            // 把处理后的缓冲内容写入Pipe
            // 注意：如果消费者没及时读Pipe，这里还是会阻塞，所以一定要保证消费者线程持续运行
            delegateSink.write(slidingBuffer.clone(), slidingBuffer.size());
        }
    }

    @Override
    public void flush() throws IOException {
        synchronized (lock) {
            delegateSink.flush();
        }
    }

    @Override
    public void close() throws IOException {
        synchronized (lock) {
            slidingBuffer.close();
            delegateSink.close();
        }
    }

    @Override
    public Timeout timeout() {
        return delegateSink.timeout();
    }
}

使用的时候这样写：

Pipe pipe = new Pipe(8192);
// 用自定义滑动窗口Sink包装Pipe的Sink
SlidingWindowSink slidingSink = new SlidingWindowSink(pipe.sink(), 8192);
BufferedSink bufferedSink = Okio.buffer(slidingSink);

// 收到音频数据时写入
bufferedSink.write(data);
bufferedSink.flush(); // 按需决定是否立即刷新

方式二：直接用线程安全的滑动缓冲替代Pipe #

如果你的需求只是在多线程间传递音频数据，其实可以不用Pipe，直接写一个线程安全的滑动缓冲类，消费者定期从中读取最新的固定大小数据就行：

import okio.Buffer;
import java.io.IOException;
import java.util.concurrent.locks.ReentrantLock;

public class AudioSlidingBuffer {
    private final Buffer buffer = new Buffer();
    private final long maxSize;
    private final ReentrantLock lock = new ReentrantLock();

    public AudioSlidingBuffer(long maxSize) {
        this.maxSize = maxSize;
    }

    public void write(byte[] data) throws IOException {
        lock.lock();
        try {
            buffer.write(data);
            if (buffer.size() > maxSize) {
                buffer.skip(buffer.size() - maxSize);
            }
        } finally {
            lock.unlock();
        }
    }

    public byte[] readLatest() throws IOException {
        lock.lock();
        try {
            int readSize = (int) Math.min(buffer.size(), maxSize);
            byte[] result = new byte[readSize];
            buffer.read(result);
            // 如果需要保留缓冲供下次读取，就不要调用buffer.clear()
            return result;
        } finally {
            lock.unlock();
        }
    }
}

• 如果坚持用Pipe，必须确保消费者线程持续读取Pipe的source，不然哪怕有滑动缓冲，最终写入Pipe时还是会阻塞。
• 音频处理基本都是多线程场景，所以所有缓冲操作一定要保证线程安全，避免数据乱掉。
• 滑动缓冲的大小要结合你的音频参数（采样率、位深、声道数）来设置，确保能覆盖你需要的预缓冲时长。

内容的提问来源于stack exchange，提问作者miibpa