java - Spring批处理条件步骤，未正确执行

Question

我基本上是 Spring Batch 的新手。第一步，我从 SFTP 服务器获取文件并使用 JSch 将其传输到本地目录。对于第二步，我读取本地目录中的文件并将内容写入数据库中。问题在于，在某些情况下，仅执行 saveWithdrawalTransaction (step2)，而不执行 getRemoteFile (step1)

配置类:

@Qualifier("writer")
@Autowired
JdbcBatchItemWriter<TopUpTransaction> writer;

@Qualifier("reader")
@Autowired
FlatFileItemReader<TopUpTransaction> reader;

@Bean(name="importWithdrawalTransactionJob")
public Job importWithdrawalTransactionJob() throws Exception {
    return jobBuilderFactory.get("importWithdrawalTransactionJob")
            .incrementer(new RunIdIncrementer())
            .start(getRemoteFile()).on(ExitStatus.FAILED.getExitCode()).end()
            .next(saveWithdrawalTransaction()).end()
            .build();
}

@Bean
public GetRemoteFileTasklet sFtpGetRemoteFilesTasklet() {
    return new GetRemoteFileTasklet();
}

@Bean
public Step getRemoteFile() throws Exception {
    return stepBuilderFactory.get("getRemoteFile")
            .tasklet(sFtpGetRemoteFilesTasklet())
            .build();
}

@Bean
public Step saveWithdrawalTransaction() throws Exception {
    return stepBuilderFactory.get("saveWithdrawalTransaction")
            .<TopUpTransaction, TopUpTransaction> chunk(1)
            .reader(reader)
            .processor(processor())
            .writer(writer)
            .build();
}

读者:

@Bean
public FlatFileItemReader<TopUpTransaction> reader() throws Exception {

File folder = new File(csvFilePath);
File filesToProcess[] = folder.listFiles();
FlatFileItemReader<TopUpTransaction> reader = new FlatFileItemReader<TopUpTransaction>();

if (filesToProcess.length > 0) {
for (File file : filesToProcess) {
    if (file.isFile()) {
        System.out.println(file.getName());
        reader.setResource( new FileSystemResource(file.getAbsolutePath()));
        reader.setLineMapper(new DefaultLineMapper<TopUpTransaction>() {{
            setLineTokenizer(new DelimitedLineTokenizer() {{
                setNames(new String[] { "lineIdentifier","transactionDate", "beneAccNum", "beneAccName", "senderName", "senderInfo",
                        "creditedAmountWithCurrency", "originalAmountWithCurrency", "netAmountWithCurrency", 
                        "charges", "otherInfo1", "otherInfo2",
                        "ubTransactionId"});
                setDelimiter("|");
            }});
            setFieldSetMapper(new CustomTopUpMapper());          
        }});
        reader.setRecordSeparatorPolicy(new CustomReaderPolicy());
        System.out.println("end read");
        file.renameTo(new File(file.getName() + "-processed"));
    }
}

作者:

@Bean(name="writer")
public JdbcBatchItemWriter<TopUpTransaction> writer() {

    StringBuffer insertWithdrawalTransaction = new StringBuffer();
    insertWithdrawalTransaction.append("INSERT INTO top_up_transactions (customer_id,txn_date, bene_acc_num, bene_acc_name,");
    insertWithdrawalTransaction.append("sender_name, sender_info, credited_amount, credited_currency,");
    insertWithdrawalTransaction.append("original_amount, original_currency,");
    insertWithdrawalTransaction.append("net_amount, net_currency, charges, other_info_1, other_info_2,");
    insertWithdrawalTransaction.append("ubp_txn_id,status) ");
    insertWithdrawalTransaction.append("VALUES (:customerId, :transactionDate,:beneAccNum,");
    insertWithdrawalTransaction.append(":beneAccName,:senderName,:senderInfo,:creditedAmount,:creditedAmountCurrency,");
    insertWithdrawalTransaction.append(":originalAmount,:originalAmountCurrency,");
    insertWithdrawalTransaction.append(":netAmount,:netAmountCurrency,:charges,:otherInfo1,:otherInfo2,");
    insertWithdrawalTransaction.append(":ubTransactionId,:status)");
    System.out.println("start write");
    JdbcBatchItemWriter<TopUpTransaction> writer = new JdbcBatchItemWriter<TopUpTransaction>();
    writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<TopUpTransaction>());
    writer.setSql(insertWithdrawalTransaction.toString());
    writer.setDataSource(dataSource);
    System.out.println("end write");

    return writer;
}

任务:

public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception
{
    Session session = null;
    Channel channel = null;
    //deleteLocalFiles();
    System.out.println("=====start tasklet=====");
    try {
        JSch ssh = new JSch();
        session = ssh.getSession("username", "host", 22);
        session.setPassword("password");
        session.setConfig("StrictHostKeyChecking", "no");
        session.connect();
        channel = session.openChannel("sftp");
        channel.connect();
        ChannelSftp sftp = (ChannelSftp) channel;
        sftp.get("*.rpt", "src/main/resources/files/");
    } catch (JSchException e) {
        e.printStackTrace();
    } catch (SftpException e) {
        e.printStackTrace();
    } finally {
        if (channel != null) {
            channel.disconnect();
        }
        if (session != null) {
            session.disconnect();
        }
    }

我不知道如何解决这个问题...我一直在尝试从各地找到的东西。请帮忙谢谢!

Answer 1

您可以使用 JobExecutionDeciderdecision 来控制失败期间的流程。

对于 Xml，你可以提到这一点

<decision id="decision" decider="decider">
            <next on="FAILED" to="step2" />
            <next on="COMPLETED" to="step3" />
        </decision>

或

jobs.get("job")
           .preventRestart()
           .listener(failedCleanupListener)
           .flow(step1)
           .next(step2)
           .next(step3)
           .end()
           .build();