逐行过滤输入流

我正在从 Amazon S3 检索大型 gzip 压缩文件。我希望能够即时转换这些文件的每一行并将输出上传到另一个 S3 存储桶。

上传 API 需要一个InputStream https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/model/PutObjectRequest.html#PutObjectRequest-java.lang.String-java.lang.String-java.io.InputStream-com.amazonaws.services.s3.model.ObjectMetadata-作为输入。

S3Object s3object = s3.fetch(bucket, key);

InputStream is = new GZIPInputStream(s3object.getObjectContent());

// . . . ?

s3.putObject(new PutObjectRequest(bucket, key, is, metadata));

我相信最有效的方法是创建我自己的自定义输入流，它将原始输入流转换为另一个输入流。我对这种方法不太熟悉，很想了解更多信息。

基本思路如下。

它的效率不是很高，但应该可以完成工作。

public class MyInputStream extends InputStream {

    private final BufferedReader input;
    private final Charset encoding = StandardCharsets.UTF_8;
    private ByteArrayInputStream buffer;

    public MyInputStream(InputStream is) throws IOException {
        input = new BufferedReader(new InputStreamReader(is, this.encoding));
        nextLine();
    }

    @Override
    public int read() throws IOException {
        if (buffer == null) {
            return -1;
        }
        int ch = buffer.read();
        if (ch == -1) {
            if (!nextLine()) {
                return -1;
            }
            return read();
        }
        return ch;
    }

    private boolean nextLine() throws IOException {
        String line;
        while ((line = input.readLine()) != null) {
            line = filterLine(line);
            if (line != null) {
                line += '\n';
                buffer = new ByteArrayInputStream(line.getBytes(encoding));
                return true;
            }
        }
        return false;
    }

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

    private String filterLine(String line) {
        // Filter the line here ... return null to skip the line
        // For example:
        return line.replace("ABC", "XYZ");
    }

}

nextLine()用（过滤后的）行预先填充行缓冲区。然后read()（由上传作业调用）从缓冲区中逐一获取字节并调用nextLine()再次加载下一行。

Use as:

s3.putObject(new PutObjectRequest(bucket, key, new MyInputStream(is), metadata));

性能改进还可以是实施int read(byte[] b, int off, int len)方法（如果 cpu 使用率很高）并使用BufferedInputStream如果 S3 客户端内部不使用缓冲区（我不知道）。