使用 Spring Batch 将文件解密到流、转换和加密为文件

Question

目标 - 解密 .pgp 加密文件，以流形式读取数据，根据供应商要求执行转换，以流形式加密并写入文件。

Logic - 自定义 Reader、Writer 和 Tasklet，将解密/加密的数据存储到 ExecutionContext 并传递到不同的步骤。

适用于 - 小文件 (~1MB)

面临的问题 - 尝试使用（~10MB - 10K 记录） - 读取步骤成功，但是当开始将数据作为加密文件写入时 - 内存问题 - java.lang.OutOfMemoryError: Java heap space

代码 sn-p -

<job id="testJob" xmlns="http://www.springframework.org/schema/batch">

    <!-- Read Encrypted file and Decrypt -->
    <batch:step id="decryptFile" next="readAndWriteData">
        <batch:tasklet ref="fileDecryptionTasklet">
            <batch:listeners>
                <batch:listener ref="decryptFileListener" />
            </batch:listeners>
        </batch:tasklet>
    </batch:step>

    <!-- Read data from decryption step and write to Stream -->
    <batch:step id="readAndWriteData" next="encryptFile">
        <batch:tasklet>
            <batch:chunk reader="hrdsCustomReader" processor="Processor"
                writer="CustomWriter" commit-interval="${.ftp.comit.interval}" />
            <batch:listeners>
                <batch:listener ref="encryptFileListener" />
            </batch:listeners>
        </batch:tasklet>
    </batch:step>

    <!-- Write to vendor specific file -->
    <batch:step id="encryptFile">
        <batch:tasklet ref="fileEncryptionTasklet" />
    </batch:step>

</job>

Tasklet 和自定义编写器代码 sn-ps -

@Override public String read() throws Exception, UnexpectedInputException, ParseException { decryptedData = (String) stepExecution.getJobExecution() .getExecutionContext().get("DecryptedData"); if (decryptedData != null) //logger.info("decryptedData in Custom Reader - \n" + decryptedData); stepExecution.getJobExecution().getExecutionContext() .put("DecryptedData", null); return decryptedData; } public void write(List items) throws Exception { logger.info("Begin writing data as an Encrypted File"); Iterator itr = items.iterator(); while(itr.hasNext()) { String element = itr.next(); lineBuffer.append(element+LINE_SEPARATOR); } ExecutionContext stepContext = this.stepExecution.getExecutionContext(); stepContext.put("EncryptedData", lineBuffer); } public RepeatStatus execute(StepContribution step, ChunkContext chunk) throws Exception { InputStream inputstream = new FileInputStream(inputdirectory); Message encryptMessage = MessageBuilder .withPayload(inputstream) .setHeader( FileEncryptionTransformer.ENCRYPTION_OPERATION_HEADER, "decryptAndVerify") .setHeader( FileEncryptionTransformer.ENCRYPTION_OPERATION_HEADER, EncryptionUtil.DECRYPT_STREAM_OPERATION) .setHeader(FileEncryptionTransformer.SOURCE_FILE_NAME_HEADER, filename).build(); InputStream inputStream = pgptransformer .doTransformStream(encryptMessage); String strData = IOUtils.toString(inputStream, "UTF-8"); inputstream.close(); chunk.getStepContext().getStepExecution().getExecutionContext().put("DecryptedData", strData); return null; } 公共重复状态执行（StepContribution 步骤，ChunkContext 块） 抛出异常 { lineBuffer = (StringBuffer) chunk.getStepContext() .getJobExecutionContext().get("EncryptedData"); byte[] bytes = lineBuffer.toString().getBytes(); InputStream inputStream = new ByteArrayInputStream(bytes); 消息 encryptMessage = MessageBuilder .withPayload(inputStream) .setHeader(PGPFileTransformer.OUTPUT_FILE_FOLDER, 输出目录） .setHeader( FileEncryptionTransformer.ENCRYPTION_OPERATION_HEADER, “签名和加密”） .setHeader( FileEncryptionTransformer.ENCRYPTION_OPERATION_HEADER, EncryptionUtil.ENCRYPT_STREAM_OPERATION) .setHeader(FileEncryptionTransformer.SOURCE_FILE_NAME_HEADER, 文件名).build(); pgptransformer.doTransform(encryptMessage); inputStream.close(); chunk.getStepContext().getStepExecution().getExecutionContext().put("EncryptedData", null); 返回空值； }

感谢有人可以帮助解决问题。

Answer 1

能够在

逻辑 - 分块处理 - 200 条记录到 1 行。

在下面发布代码 -

批量配置-

<job id="aic-batch-xxx-ftp" xmlns="http://www.springframework.org/schema/batch">


    <!-- Read data , decrypt , process and write to encrypted file -->
    <batch:step id="readAndWriteData">
        <batch:tasklet>
            <batch:chunk reader="xxxCustomReader" processor="xxxFileProccessor"
                writer="xxxCustomWriter" commit-interval="${aic.batch.xxx.ftp.comit.interval}" />
        </batch:tasklet>
    </batch:step>


</job>

阅读器逻辑 -

StringBuffer decryptedData = new StringBuffer();
    String strLine = "";

    PGPLib pgp = new PGPLib();
    KeyStore keyStore = new KeyStore("xxx.keystore", "xxx");

    long startTime = System.currentTimeMillis();
    // Read & decrypt File Line By Line
    if ((strLine = bufferedReader.readLine()) != null) {
        strLine = strLine.replace("NEW_LINE", "\r\n");
        decryptedData.append((pgp.decryptString(strLine, keyStore,
                "xxx")));
        long endTime = System.currentTimeMillis();
        logger.debug("Total time taken = " + (endTime - startTime) + " msec");
        return decryptedData;
    }
    else
        return null;

作家逻辑 -

public void write(List<? extends StringBuffer> items) throws Exception {
    logger.debug("Begin writing data as an Encrypted File");

    @SuppressWarnings("unchecked")
    Iterator<StringBuffer> itr = (Iterator<StringBuffer>) items.iterator();
    while (itr.hasNext()) {
        StringBuffer element = itr.next();
        encrypt(element);
        count++;
    }
}

public void encrypt(StringBuffer element) throws PGPException, IOException {

    PGPLib pgp = new PGPLib();
    KeyStore keyStore = new KeyStore("xxx.keystore", "xxx");

    String strLine = element.toString();
    StringBuffer buffer = new StringBuffer("");
    int i = 0;
    long startTime = System.currentTimeMillis();

    if (i % 200 == 0) {
        if (i != 0) {
            String encryptString = pgp.encryptString(buffer.toString(),
                    keyStore,
                    "xxx");
            encryptString = encryptString.replace("\r\n", "NEW_LINE");
            bufferedWriter.write(encryptString);
            bufferedWriter.newLine();
        }
        buffer = new StringBuffer(strLine);
    } else {
        buffer.append("\r\n").append(strLine);
    }
    i++;
    if (buffer != null && buffer.length() > 0) {
        String encryptString = pgp.encryptString(buffer.toString(),
                keyStore, "xxx");
        encryptString = encryptString.replace("\r\n", "NEW_LINE");
        bufferedWriter.write(encryptString);
        bufferedWriter.newLine();
    }

    long endTime = System.currentTimeMillis();
    logger.debug("Total time taken = " + (endTime - startTime) + " msec");

}