如何为 jdbc 使用复合文件编写器或将数据写入数据库？答案

【问题标题】：How to use composite file writer for jdbc or writing data to database?如何为 jdbc 使用复合文件编写器或将数据写入数据库？
【发布时间】：2022-01-23 14:28:16
【问题描述】：

我已经编写了一个用于写入平面文件的复合项目编写器，但如果满足某些条件，我也想在数据库中写入。那么，我该怎么做……我是初学者。请帮忙..

复合文件写入器类：

@Component("CompositeFileWriter")

公共类 ChargeBackCompositeFileWriter 实现 ItemStreamWriter{

@Autowired
private PartOrderDao partOrderDao;

@Autowired
@Qualifier("errorFileWriter")
private ItemStreamWriter<ChargeBack> errorFileWriter;

@Autowired
@Qualifier("retryFileWriter")
private ItemStreamWriter<ChargeBack> retryFileWriter;



@Autowired
@Qualifier("processedFileWriter")
private ItemStreamWriter<ChargeBack> processedFileWriter;




@Override
public void write(List<? extends ChargeBack> items) throws Exception {
    
    
    
    List<ChargeBack> errorRecords = items.stream().filter(ChargeBack::isBadRecord).collect(Collectors.toList());

    List<ChargeBack> retryRecords = items.stream().filter(ChargeBack::isRetryRecord)
            .collect(Collectors.toList());
    
    List<ChargeBack> processedRecords = items.stream()
            .filter(chargeBack -> !chargeBack.isBadRecord())
            .filter(chargeBack-> !chargeBack.isRetryRecord())
            .collect(Collectors.toList());
    
    
    
    
    //ChargeBack chargeBack = new ChargeBack();

    if (CollectionUtils.isNotEmpty(errorRecords)) {
        errorFileWriter.write(errorRecords);
    }

    if (CollectionUtils.isNotEmpty(processedRecords)) {
        processedFileWriter.write(processedRecords);
    }
    
    if (CollectionUtils.isNotEmpty(retryRecords)) {
        
    }
}


@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
    errorFileWriter.open(executionContext);
    processedFileWriter.open(executionContext);
    //retryFileWriter.open(executionContext);
}

@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
    errorFileWriter.update(executionContext);
    processedFileWriter.update(executionContext);
    //retryFileWriter.update(executionContext);
}

@Override
public void close() throws ItemStreamException {
    errorFileWriter.close();
    processedFileWriter.close();
    //retryFileWriter.close();
}

}

批处理配置文件

@Bean
public ItemStreamWriter<ChargeBack> processedFileWriter(
        @Value("${processed.file}") FileSystemResource processedFile) {
    FlatFileItemWriter<ChargeBack> writer = new FlatFileItemWriter<>();
    writer.setResource(processedFile);
    writer.setLineAggregator(new DelimitedLineAggregator<ChargeBack>() {
        {
            setDelimiter("|");
            setFieldExtractor(new BeanWrapperFieldExtractor<ChargeBack>() {
                {
                    setNames(new String[] { ""
                              });
                }
            });
        }
    });
    return writer;
}

//for reading retry file and passing it, for writing record to the table.
@Bean
@StepScope
public FlatFileItemReader retryFileReader(@Value("${input.retryFile}") FileSystemResource inputFile){
    FlatFileItemReader reader = new FlatFileItemReader();
    reader.setResource(inputFile);
    reader.setLinesToSkip(0);
    reader.setStrict(false);
    reader.setLineMapper(new DefaultLineMapper(){
        {setFieldSetMapper(new BeanWrapperFieldSetMapper(){{
         setTargetType(ChargeBack.class);}});
         setLineTokenizer(new DelimitedLineTokenizer(){{
         setNames(new String[]{""});
         setDelimiter("|");}});}});
    return reader;
}


 @Bean
 public JdbcBatchItemWriter retryFileWriterToDB(){
     JdbcBatchItemWriter writer = new JdbcBatchItemWriter ();
     writer.setDataSource(pdccRepositoryConfig.pdccDataSource());
     writer.setSql(")" +
             " values (?, ?, ?, ? , ?, ?, ? , ?, ?, ? , ?, ?, ?, ?, ? ) ");
     writer.setItemPreparedStatementSetter(new ItemPreparedStatementSetter<ChargeBack>() {
         @Override
         public void setValues(ChargeBack item, PreparedStatement ps) throws SQLException {
             ps.setString(1, item.getAccertifyCBControlNumber().toString());
             ps.setString(2,item.getInternalComment2());
             ps.setString(3, item.getDisputeType());
             ps.setString(4, item.getBrand());
             ps.setString(5,item.getResolution1ts());
             ps.setString(6, item.getReasonCode());
             ps.setString(7, item.getReasonDescription());
             ps.setString(8,item.getWonLossStatus());
             ps.setString(9, item.getDisputedDate());
             ps.setString(10, Double.toString(item.getDisputedAmount()));
             ps.setString(11, item.getReportingGroup());
             ps.setString(12, item.getDueDate());
             ps.setString(13,item.getLocation());
             ps.setString(14, item.getResolution1ts());
             ps.setString(15, item.getTransactionDate());
            
         }

        
     });

     return writer;

 }

我目前正在使用上面的 jdbcbatchitemwriter 在数据库中单独编写，但我希望它被 streamitemwriter 使用，这样我就可以同时编写，而不是单独编写。我的问题是我想使用这个 jdbcbatchitemwriter 供 streamitemwriter 使用，就像上面我使用平面文件项目编写器一样。如果我不能在 streamwriter 中使用 jdbcbatchitemwriter 那么下一个解决方案是什么。如果可能，请详细说明..

【问题讨论】：

标签： java spring spring-boot spring-batch

【解决方案1】：

所以，要将 JDBCitemwriter 与 Compositeitemwriter 一起使用，只需将 JdbcitemWriter 返回类型设置为 itemwriter。

 @Bean 
 public ItemWriter retryFileWriterToDB(){
  JdbcBatchItemWriter writer = new JdbcBatchItemWriter ();
     writer.setDataSource(pdccRepositoryConfig.pdccDataSource());
     writer.setSql("insert into failed_forter_claim_api_order (accertify_cb_control_number, internal_comment2, dispute_type, brand, resolution1_name, reason_code, reason_description, won_loss_status, disputed_date, disputed_amount, reporting_group, due_date, location, resolution1_ts, transaction_date)" +
             " values (?, ?, ?, ? , ?, ?, ? , ?, ?, ? , ?, ?, ?, ?, ? ) ");
     writer.setItemPreparedStatementSetter(new ItemPreparedStatementSetter<ChargeBack>() {
         @Override
         public void setValues(ChargeBack item, PreparedStatement ps) throws SQLException {
             ps.setString(1, item.getAccertifyCBControlNumber().toString());
             ps.setString(2,item.getInternalComment2());
             ps.setString(3, item.getDisputeType());
             ps.setString(4, item.getBrand());
             ps.setString(5,item.getResolution1ts());
             ps.setString(6, item.getReasonCode());
             ps.setString(7, item.getReasonDescription());
             ps.setString(8,item.getWonLossStatus());
             ps.setString(9, item.getDisputedDate());
             ps.setString(10, Double.toString(item.getDisputedAmount()));
             ps.setString(11, item.getReportingGroup());
             ps.setString(12, item.getDueDate());
             ps.setString(13,item.getLocation());
             ps.setString(14, item.getResolution1ts());
             ps.setString(15, item.getTransactionDate());
            
         }

        
     });

     return writer;

 }

在您的compositeItemwriter 中自动装配，如下所示，您可以更喜欢retryFileWriterToDB 作为关键字，我是如何在下面的代码中使用它的。

@Component("CompositeFileWriter") 公共类 ChargeBackCompositeFileWriter 实现 ItemStreamWriter{

@Autowired
private PartOrderDao partOrderDao;

@Autowired
@Qualifier("errorFileWriter")
private ItemStreamWriter<ChargeBack> errorFileWriter;

@Autowired
@Qualifier("retryFileWriterToDB")
private ItemWriter<ChargeBack> retryFileWriterToDB;



@Autowired
@Qualifier("processedFileWriter")
private ItemStreamWriter<ChargeBack> processedFileWriter;




@Override
public void write(List<? extends ChargeBack> items) throws Exception {
    

    
    List<ChargeBack> errorRecords = items.stream().filter(ChargeBack::isBadRecord).collect(Collectors.toList());

    List<ChargeBack> retryRecords = items.stream().filter(ChargeBack::isRetryRecord)
            .collect(Collectors.toList());
    
    List<ChargeBack> processedRecords = items.stream()
            .filter(chargeBack -> !chargeBack.isBadRecord())
            .filter(chargeBack-> !chargeBack.isRetryRecord())
            .collect(Collectors.toList());
    
    
    
    
    //ChargeBack chargeBack = new ChargeBack();

    if (CollectionUtils.isNotEmpty(errorRecords)) {
        errorFileWriter.write(errorRecords);
    }

    if (CollectionUtils.isNotEmpty(processedRecords)) {
        processedFileWriter.write(processedRecords);

    }
    
    if (CollectionUtils.isNotEmpty(retryRecords)) {
        
        System.out.println("writing ti DB:..");
    //  System.out.println("Calling partdao.......");
        //partOrderDao.insertFailedRecord(retryRecords);
        retryFileWriterToDB.write(retryRecords);
    }
}


@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
    errorFileWriter.open(executionContext);
    processedFileWriter.open(executionContext);
    //((ItemStream) retryFileWriter).open(executionContext);
}

@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
    errorFileWriter.update(executionContext);
    processedFileWriter.update(executionContext);
    //((ItemStream) retryFileWriter).update(executionContext);
}

@Override
public void close() throws ItemStreamException {
    errorFileWriter.close();
    processedFileWriter.close();
    //((ItemStream) retryFileWriter).close();
}

}

其余所有代码都与粘贴的问题相同。在 batchCONfig 文件中，我只是更改了 jdbcitemwriter 的返回类型，在 Compositewriter 类中，我已经自动装配了这个 jdbcitemwriter，就像这样 @Autowired @Qualifier("retryFileWriterToDB") private ItemWriter<ChargeBack> retryFileWriterToDB; 并在复合编写器类中使用它作为

        retryFileWriterToDB.write(retryRecords);

【讨论】：