缓冲后台InputStream实现

Question

我编写了包装其他流的后台InputStream(和OutputStream)实现,并在后台线程上预读,主要允许解压缩/压缩在解压缩流处理的不同线程中发生.

这是一个相当标准的生产者/消费者模型.

这似乎是一种简单的方法,可以通过简单的读取,处理和写入数据的流程来充分利用多核CPU,从而更有效地利用CPU和磁盘资源.也许"高效"并不是最好的词,但它提供了更高的利用率,对我来说更感兴趣,减少了运行时间,而不是直接从a读取ZipInputStream并直接写入ZipOutputStream.

我很高兴发布代码,但我的问题是我是否正在重新发明现有(和更多运行)库中现有的东西？

编辑 - 发布代码......

我的代码BackgroundInputStream是在下面(BackgroundOutputStream非常相似),但有一些方面我想改进.

1. 看起来我工作得太过艰难,无法前后传递缓冲区.
2. 如果调用代码抛弃了对引用的引用BackgroundInputStream,那么backgroundReaderThread它将永远存在.
3. 信号eof需要改进.
4. 应将异常传播到前台线程.
5. 我想允许使用提供的线程Executor.
6. 该close()方法应该通知后台线程,并且不应该关闭包装的流,因为包装的流应该由从其读取的后台线程拥有.
7. 在收盘后做一些愚蠢的事情应该适当地照顾.

---

package nz.co.datacute.io;

import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.concurrent.LinkedBlockingQueue;

public class BackgroundInputStream extends InputStream {
    private static final int DEFAULT_QUEUE_SIZE = 1;
    private static final int DEFAULT_BUFFER_SIZE = 64*1024;
    private final int queueSize;
    private final int bufferSize;
    private volatile boolean eof = false;
    private LinkedBlockingQueue<byte[]> bufferQueue;
    private final InputStream wrappedInputStream;
    private byte[] currentBuffer;
    private volatile byte[] freeBuffer;
    private int pos;

    public BackgroundInputStream(InputStream wrappedInputStream) {
        this(wrappedInputStream, DEFAULT_QUEUE_SIZE, DEFAULT_BUFFER_SIZE);
    }

    public BackgroundInputStream(InputStream wrappedInputStream,int queueSize,int bufferSize) {
        this.wrappedInputStream = wrappedInputStream;
        this.queueSize = queueSize;
        this.bufferSize = bufferSize;
    }

    @Override
    public int read() throws IOException {
        if (bufferQueue == null) {
            bufferQueue = new LinkedBlockingQueue<byte[]>(queueSize);
            BackgroundReader backgroundReader = new BackgroundReader();
            Thread backgroundReaderThread = new Thread(backgroundReader, "Background InputStream");
            backgroundReaderThread.start();
        }
        if (currentBuffer == null) {
            try {
                if ((!eof) || (bufferQueue.size() > 0)) {
                    currentBuffer = bufferQueue.take();
                    pos = 0;
                } else {
                    return -1;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        int b = currentBuffer[pos++];
        if (pos == currentBuffer.length) {
            freeBuffer = currentBuffer;
            currentBuffer = null;
        }
        return b;
    }

    @Override
    public int available() throws IOException {
        if (currentBuffer == null) return 0;
        return currentBuffer.length;
    }

    @Override
    public void close() throws IOException {
        wrappedInputStream.close();
        currentBuffer = null;
        freeBuffer = null;
    }

    class BackgroundReader implements Runnable {

        @Override
        public void run() {
            try {
                while (!eof) {
                    byte[] newBuffer;
                    if (freeBuffer != null) {
                        newBuffer = freeBuffer;
                        freeBuffer = null;
                    } else {
                        newBuffer = new byte[bufferSize];
                    }
                    int bytesRead = 0;
                    int writtenToBuffer = 0;
                    while (((bytesRead = wrappedInputStream.read(newBuffer, writtenToBuffer, bufferSize - writtenToBuffer)) != -1) && (writtenToBuffer < bufferSize)) {
                        writtenToBuffer += bytesRead;
                    }
                    if (writtenToBuffer > 0) {
                        if (writtenToBuffer < bufferSize) {
                            newBuffer = Arrays.copyOf(newBuffer, writtenToBuffer);
                        }
                        bufferQueue.put(newBuffer);
                    }
                    if (bytesRead == -1) {
                        eof = true;
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}

Answer 1

听起来不错。我从来没有遇到过任何开箱即用的东西，但尝试使用空闲核心进行压缩（如果可用）是非常有意义的。

也许您可以利用Commons I/O - 这是一个经过良好测试的库，可以帮助处理一些更无聊的东西，让您专注于扩展很酷的并行部分。也许您甚至可以将您的代码贡献给 Commons 项目;-)