在 Java 中将文件的前 N ​​个字节作为输入流读取？

在我的一生中，我一直无法找到与我想做的事情相匹配的问题，所以我将在这里解释我的用例。如果您知道某个主题已经涵盖了该问题的答案，请随时引导我找到该主题。 :)

我有一段代码可以定期（每 20 秒）将文件上传到 Amazon S3。该文件是由另一个进程写入的日志文件，因此此函数实际上是一种跟踪日志的方法，以便某人可以半实时读取其内容，而无需直接访问日志所在的计算机。

直到最近，我只是使用 S3 PutObject 方法（使用文件作为输入）来执行此上传。但在 AWS SDK 1.9 中，这不再有效，因为如果实际上传的内容大小大于上传开始时承诺的内容长度，S3 客户端会拒绝请求。此方法在开始流式传输数据之前读取文件的大小，因此考虑到此应用程序的性质，文件的大小很可能在该点和流结束之间增加。这意味着我现在需要确保只发送 N 字节的数据，无论文件有多大。

我不需要以任何方式解释文件中的字节，所以我不关心编码。我可以逐字节传输它。基本上，我想要的是一种简单的方法，我可以读取文件直至第 N 个字节，然后让它终止读取，即使文件中还有更多数据超过该点。 （换句话说，将 EOF 插入流中的特定点。）

例如，如果我的文件在开始上传时为 10000 字节长，但在上传过程中增长到 12000 字节，则无论大小如何变化，我都希望在 10000 字节时停止上传。 （在后续上传时，我会上传 12000 字节或更多。）

我还没有找到一种预先制作的方法来做到这一点 - 到目前为止我发现的最好的方法似乎是 IOUtils.copyLarge(InputStream, OutputStream, offset, length)，它可以被告知复制最大“长度”字节到提供的 OutputStream。然而，copyLarge 是一种阻塞方法，PutObject 也是如此（它可能在其 InputStream 上调用一种形式的 read()），所以看来我根本无法让它工作。

我还没有找到任何方法或预构建的流可以做到这一点，所以这让我觉得我需要编写自己的实现来直接监视已读取的字节数。这可能会像 BufferedInputStream 一样工作，其中每批读取的字节数是缓冲区大小或要读取的剩余字节中的较小者。 （例如，如果缓冲区大小为 3000 字节，我会执行三个批次，每个批次为 3000 字节，然后是一个批次为 1000 字节 + EOF。）

有谁知道更好的方法来做到这一点？谢谢。

EDIT只是为了澄清一下，我已经知道了几种替代方案，但它们都不理想：

(1)我可以在上传文件时锁定文件。这样做会导致写入文件的过程中数据丢失或操作问题。

(2) 我可以在上传文件之前创建文件的本地副本。这可能效率非常低，并且占用大量不必要的磁盘空间（该文件可能会增长到几 GB 范围，并且运行它的计算机可能磁盘空间不足）。

EDIT 2:根据同事的建议，我的最终解决方案如下所示：

private void uploadLogFile(final File logFile) {
    if (logFile.exists()) {
        long byteLength = logFile.length();
        try (
            FileInputStream fileStream = new FileInputStream(logFile);
            InputStream limitStream = ByteStreams.limit(fileStream, byteLength);
        ) {
            ObjectMetadata md = new ObjectMetadata();
            md.setContentLength(byteLength);
            // Set other metadata as appropriate.
            PutObjectRequest req = new PutObjectRequest(bucket, key, limitStream, md);
            s3Client.putObject(req);
        } // plus exception handling
    }
}

LimitInputStream 是我的同事建议的，显然不知道它已被弃用。 ByteStreams.limit 是当前 Guava 的替代品，它可以满足我的需求。感谢大家。

完整答案删除和替换：

包裹起来相对简单InputStream例如在发出数据结束信号之前限制其将传递的字节数。FilterInputStream是针对这种一般类型的工作，但是由于您必须为此覆盖几乎所有方法特别的工作，它只是妨碍。

这是一个粗略的解决方案：

import java.io.IOException;
import java.io.InputStream;

/**
 * An {@code InputStream} wrapper that provides up to a maximum number of
 * bytes from the underlying stream.  Does not support mark/reset, even
 * when the wrapped stream does, and does not perform any buffering.
 */
public class BoundedInputStream extends InputStream {

    /** This stream's underlying @{code InputStream} */
    private final InputStream data;

    /** The maximum number of bytes still available from this stream */ 
    private long bytesRemaining;

    /**
     * Initializes a new {@code BoundedInputStream} with the specified
     * underlying stream and byte limit
     * @param data the @{code InputStream} serving as the source of this
     *        one's data
     * @param maxBytes the maximum number of bytes this stream will deliver
     *        before signaling end-of-data
     */
    public BoundedInputStream(InputStream data, long maxBytes) {
        this.data = data;
        bytesRemaining = Math.max(maxBytes, 0);
    }

    @Override
    public int available() throws IOException {
        return (int) Math.min(data.available(), bytesRemaining);
    }

    @Override
    public void close() throws IOException {
        data.close();
    }

    @Override
    public synchronized void mark(int limit) {
        // does nothing
    }

    @Override
    public boolean markSupported() {
        return false;
    }

    @Override
    public int read(byte[] buf, int off, int len) throws IOException {
        if (bytesRemaining > 0) {
            int nRead = data.read(
                    buf, off, (int) Math.min(len, bytesRemaining));

            bytesRemaining -= nRead;

            return nRead;
        } else {
            return -1;
        }
    }

    @Override
    public int read(byte[] buf) throws IOException {
        return this.read(buf, 0, buf.length);
    }

    @Override
    public synchronized void reset() throws IOException {
        throw new IOException("reset() not supported");
    }

    @Override
    public long skip(long n) throws IOException {
        long skipped = data.skip(Math.min(n, bytesRemaining));

        bytesRemaining -= skipped;

        return skipped;
    }

    @Override
    public int read() throws IOException {
        if (bytesRemaining > 0) {
            int c = data.read();

            if (c >= 0) {
                bytesRemaining -= 1;
            }

            return c;
        } else {
            return -1;
        }
    }
}