基于Spring Batch 配置重试逻辑

sammy, 1234 , 31 / 10 / 2015 , 10000

john, 9999 , 3 / 12 / 2015 , 12321

@XmlRootElement (name = [transactionRecord])

@Data

public class Transaction {

private String username;

private int userId;

private int age;

private String postCode;

private LocalDateTime transactionDate;

private double amount;

}

public class RetryItemProcessor implements ItemProcessor<Transaction, Transaction> {

     private static final Logger LOGGER = LoggerFactory.getLogger(RetryItemProcessor. class );

     @Autowired

     private CloseableHttpClient closeableHttpClient;

     @Override

     public Transaction process(Transaction transaction) throws IOException, JSONException {

         LOGGER.info( "Attempting to process user with id={}" , transaction.getUserId());

         HttpResponse response = fetchMoreUserDetails(transaction.getUserId());

         //parse user's age and postCode from response and update transaction

         String result = EntityUtils.toString(response.getEntity());

         JSONObject userObject = new JSONObject(result);

         transaction.setAge(Integer.parseInt(userObject.getString( "age" )));

         transaction.setPostCode(userObject.getString( "postCode" ));

         return transaction;

     }

     private HttpResponse fetchMoreUserDetails( int id) throws IOException {

         final HttpGet request = new HttpGet( "http://HdhCmsTesttestapi测试数据:81/user/" + id);

         return closeableHttpClient.execute(request);

     }

}

< transactionRecord >

     < transactionRecord >

         < amount >10000.0</ amount >

         < transactionDate >2015-10-31 0</ transactionDate >

         < userId >1234</ userId >

         < username >sammy</ username >

         < age >10</ age >

         < postCode >430222</ postCode >

     </ transactionRecord >

     ...

</ transactionRecord >

@Configuration

@EnableBatchProcessing

public class SpringBatchRetryConfig {

     private static final String[] tokens = { "username" , "userid" , "transactiondate" , "amount" };

     private static final int TWO_SECONDS = 2000 ;

     @Autowired

     private JobBuilderFactory jobBuilderFactory;

     @Autowired

     private StepBuilderFactory stepBuilderFactory;

     @Value ( "input/recordRetry.csv" )

     private Resource inputCsv;

     @Value ( "file:xml/retryOutput.xml" )

     private Resource outputXml;

     public ItemReader<Transaction> itemReader(Resource inputData) throws ParseException {

         DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();

         tokenizer.setNames(tokens);

         DefaultLineMapper<Transaction> lineMapper = new DefaultLineMapper<>();

         lineMapper.setLineTokenizer(tokenizer);

         lineMapper.setFieldSetMapper( new RecordFieldSetMapper());

         FlatFileItemReader<Transaction> reader = new FlatFileItemReader<>();

         reader.setResource(inputData);

         reader.setLinesToSkip( 1 );

         reader.setLineMapper(lineMapper);

         return reader;

     }

     @Bean

     public CloseableHttpClient closeableHttpClient() {

         final RequestConfig config = RequestConfig.custom()

           .setConnectTimeout(TWO_SECONDS)

           .build();

         return HttpClientBuilder.create().setDefaultRequestConfig(config).build();

     }

     @Bean

     public ItemProcessor<Transaction, Transaction> retryItemProcessor() {

         return new RetryItemProcessor();

     }

     @Bean

     public ItemWriter<Transaction> itemWriter(Marshaller marshaller) {

         StaxEventItemWriter<Transaction> itemWriter = new StaxEventItemWriter<>();

         itemWriter.setMarshaller(marshaller);

         itemWriter.setRootTagName( "transactionRecord" );

         itemWriter.setResource(outputXml);

         return itemWriter;

     }

     @Bean

     public Marshaller marshaller() {

         Jaxb2Marshaller marshaller = new Jaxb2Marshaller();

         marshaller.setClassesToBeBound(Transaction. class );

         return marshaller;

     }

     @Bean

     public Step retryStep( @Qualifier ( "retryItemProcessor" ) ItemProcessor<Transaction, Transaction> processor,

       ItemWriter<Transaction> writer) throws ParseException {

         return stepBuilderFactory.get( "retryStep" )

           .<Transaction, Transaction>chunk( 10 )

           .reader(itemReader(inputCsv))

           .processor(processor)

           .writer(writer)

           .faultTolerant()

           .retryLimit( 3 )

           .retry(ConnectTimeoutException. class )

           .retry(DeadlockLoserDataAccessException. class )

           .build();

     }

     @Bean (name = "retryBatchJob" )

     public Job retryJob( @Qualifier ( "retryStep" ) Step retryStep) {

         return jobBuilderFactory

           .get( "retryBatchJob" )

           .start(retryStep)

           .build();

     }

@RunWith (SpringRunner. class )

@SpringBatchTest

@EnableAutoConfiguration

@ContextConfiguration (classes = { SpringBatchRetryConfig. class })

public class SpringBatchRetryIntegrationTest {

     private static final String TEST_OUTPUT = "xml/retryOutput.xml" ;

     private static final String EXPECTED_OUTPUT = "src/test/resources/output/batchRetry/retryOutput.xml" ;

     @Autowired

     private JobLauncherTestUtils jobLauncherTestUtils;

     @MockBean

     private CloseableHttpClient closeableHttpClient;

     @Mock

     private CloseableHttpResponse httpResponse;

     @Test

     public void whenEndpointAlwaysFail_thenJobFails() throws Exception {

         when(closeableHttpClient.execute(any()))

           .thenThrow( new ConnectTimeoutException( "Endpoint is down" ));

         JobExecution jobExecution = jobLauncherTestUtils.launchJob(defaultJobParameters());

         JobInstance actualJobInstance = jobExecution.getJobInstance();

         ExitStatus actualJobExitStatus = jobExecution.getExitStatus();

         assertThat(actualJobInstance.getJobName(), is( "retryBatchJob" ));

         assertThat(actualJobExitStatus.getExitCode(), is( "FAILED" ));

         assertThat(actualJobExitStatus.getExitDescription(), containsString( "org.apache.http.conn.ConnectTimeoutException" ));

     }

     @Test

     public void whenEndpointFailsTwicePasses3rdTime_thenSuccess() throws Exception {

         FileSystemResource expectedResult = new FileSystemResource(EXPECTED_OUTPUT);

         FileSystemResource actualResult = new FileSystemResource(TEST_OUTPUT);

         //前两次调用失败，第三次继续执行

         when(httpResponse.getEntity())

           .thenReturn( new StringEntity( "{ \"age\":10, \"postCode\":\"430222\" }" ));

         when(closeableHttpClient.execute(any()))

           .thenThrow( new ConnectTimeoutException( "Timeout count 1" ))

           .thenThrow( new ConnectTimeoutException( "Timeout count 2" ))

           .thenReturn(httpResponse);

         JobExecution jobExecution = jobLauncherTestUtils.launchJob(defaultJobParameters());

         JobInstance actualJobInstance = jobExecution.getJobInstance();

         ExitStatus actualJobExitStatus = jobExecution.getExitStatus();

         assertThat(actualJobInstance.getJobName(), is( "retryBatchJob" ));

         assertThat(actualJobExitStatus.getExitCode(), is( "COMPLETED" ));

         AssertFile.assertFileEquals(expectedResult, actualResult);

     }

     private JobParameters defaultJobParameters() {

         JobParametersBuilder paramsBuilder = new JobParametersBuilder();

         paramsBuilder.addString( "jobID" , String.valueOf(System.currentTimeMillis()));

         return paramsBuilder.toJobParameters();

     }

}