spring batch + spring boot 配置

1. spring batch 批处理配置

import java.io.IOException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.LineMapper;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;

/**
* spring batch 配置
* @author
*/
@Configuration
@EnableBatchProcessing
public class BlackListBatchConfiguration {

private static final Logger logger = LoggerFactory.getLogger(BlackListBatchConfiguration.class);
/**
* 读取外部文件方法
* @return
* @throws IOException
*/
@Bean
@StepScope
public ItemReader<BlackListDO> reader(@Value("#{jobParameters[inputFileBlack]}") String inputFile) throws IOException {
logger.info("inputFile:"+new ClassPathResource(inputFile).getURL().getPath());
if(inputFile == null){
logger.error("The blacklist reader file is null");
return null;
}
FlatFileItemReader<BlackListDO> reader = new FlatFileItemReader<BlackListDO>();

reader.setResource(new ClassPathResource(inputFile));

reader.setLineMapper(lineMapper());

reader.setLinesToSkip(1);

reader.open(JobCompletionNotificationListener.jobExecution.getExecutionContext());

return reader;
}

/**
* 读取文本行映射POJO
* @return
*/
@Bean
@StepScope
public LineMapper<BlackListDO> lineMapper() {
DefaultLineMapper<BlackListDO> lineMapper = new DefaultLineMapper<BlackListDO>();
DelimitedLineTokenizer lineTokenizer = new DelimitedLineTokenizer();
lineTokenizer.setDelimiter(",");
lineTokenizer.setStrict(false);
lineTokenizer.setNames(new String[] { "type","value","fraudType"});

BeanWrapperFieldSetMapper<BlackListDO> fieldSetMapper = new BeanWrapperFieldSetMapper<BlackListDO>();
fieldSetMapper.setTargetType(BlackListDO.class);
lineMapper.setLineTokenizer(lineTokenizer);
lineMapper.setFieldSetMapper(new BlackListFieldSetMapper());
return lineMapper;
}

/**
* 处理过程
* @return
*/
@Bean
@StepScope
public ItemProcessor<BlackListDO, BlackListDO> processor(@Value("#{jobParameters[inputFileBlack]}") String inputFile) {
return new BlackListDOItemProcessor(inputFile);
}

/**
* 写出内容
* @return
*/
@Bean
@StepScope
public ItemWriter<BlackListDO> writer() {
return new BlackListItemWriter();
}

/**
* 构建job
* @param jobs
* @param s1
* @param listener
* @return
*/
@Bean
public Job importFileJob(JobBuilderFactory jobs, Step step1,JobExecutionListener listener,JobRepository jobRepository) {
return jobs.get("importFileJob")
.incrementer(new RunIdIncrementer())
.repository(jobRepository)
.listener(listener)
.flow(step1)
.end()
.build();
}

/**
* 声明step
* @param stepBuilderFactory
* @param reader
* @param writer
* @param processor
* @return
*/
@Bean
public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<BlackListDO> reader,
ItemWriter<BlackListDO> writer, ItemProcessor<BlackListDO, BlackListDO> processor,PlatformTransactionManager transactionManager) {
logger.error("step1");
return stepBuilderFactory.get("step1")
.<BlackListDO, BlackListDO> chunk(500)
.reader(reader)
.processor(processor)
.writer(writer)
.faultTolerant()
.retry(Exception.class)   // 重试
.noRetry(ParseException.class)
.retryLimit(1)           //每条记录重试一次
.listener(new RetryFailuireItemListener())
.skip(Exception.class)
.skipLimit(500)         //一共允许跳过200次异常
.taskExecutor(new SimpleAsyncTaskExecutor()) //设置并发方式执行
.throttleLimit(10)        //并发任务数为 10,默认为4
.transactionManager(transactionManager)
.build();
}

}

BlackListDOItemProcessor 处理类

import java.sql.Timestamp;
import java.util.Date;
import java.util.UUID;

import org.springframework.batch.item.ItemProcessor;

import com.BlackListDO;

/**
* @author zhengyong
*
*/

public class BlackListDOItemProcessor implements ItemProcessor<BlackListDO, BlackListDO> {

public String inputFile;

public BlackListDOItemProcessor() {
}

public BlackListDOItemProcessor(String inputFile) {
this.inputFile = inputFile;
}
// 数据处理
public BlackListDO process(BlackListDO blackListDO) throws Exception {
blackListDO.setDeleteFlag(0);
blackListDO.setUuid(UUID.randomUUID().toString().replaceAll("-", ""));
return blackListDO;
}

}

BlackListItemWriter 写入数据库类

import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;

import com.BlackListDO;

public class BlackListItemWriter implements ItemWriter<BlackListDO> {

@Override
public void write(List<? extends BlackListDO> blackList) throws Exception {
// 插入数据库操作
}

}

JobCompletionNotificationListener 监听任务

import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.listener.JobExecutionListenerSupport;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class JobCompletionNotificationListener extends JobExecutionListenerSupport {

@Override
public void afterJob(JobExecution jobExecution) {

}

@Override
public void beforeJob(JobExecution jobExecution) {
super.beforeJob(jobExecution);
}
}

RetryFailuireItemListener 重试监听

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;

public class RetryFailuireItemListener implements RetryListener{

private static final Logger logger = LoggerFactory.getLogger(RetryFailuireItemListener.class);

@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
return true;
}

@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
}

@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
logger.error("【重试异常】："+throwable.getMessage());
}

}

2. spring boot 配置

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.context.web.SpringBootServletInitializer;
import org.springframework.context.annotation.ComponentScan;

@ComponentScan("com.syncclient")
@SpringBootApplication
public class SpringBootJspApplication extends SpringBootServletInitializer{

/**
* 500一批
* oracle : 单条插入基本每分钟2.5W条（50W，19.5min） ,批量插入基本每分钟10W条(50W，4.7mim)
* mysql  : 单条插入基本每分钟2.5W条（50W，11.4min） ,批量插入基本每分钟40W条(50W，1.3min)
*/
@Override
protected SpringApplicationBuilder configure(SpringApplicationBuilder application) {
return application.sources(SpringBootJspApplication.class);
}

public static void main(String[] args) throws Exception {
SpringApplication.run(SpringBootJspApplication.class,new String[]{"appStart=true"});
}
}

3. 配置文件

# mysql config
#spring.boot.database = mysql
#spring.datasource.url = jdbc:mysql://127.0.0.1:3306/spring_batch?useUnicode=true&characterEncoding=utf8
#spring.datasource.username = admin
#spring.datasource.password = 123456
#spring.datasource.driverClassName = com.mysql.jdbc.Driver
#spring.batch.schema = classpath:/org/springframework/batch/core/schema-mysql.sql
#spring.batch.drop = classpath:/org/springframework/batch/core/schema-drop-mysql.sql

# oracle config
spring.boot.database = oracle
spring.datasource.url = jdbc:oracle:thin:@127.0.0.1:1521:spring_batch
spring.datasource.username = admin
spring.datasource.password = 123456
spring.datasource.driverClassName = oracle.jdbc.driver.OracleDriver
spring.batch.schema = classpath:org/springframework/batch/core/schema-oracle10g.sql
spring.batch.drop = classpath:org/springframework/batch/core/schema-drop-oracle10g.sql

# batch config
spring.batch.job.names = importFileJob
spring.batch.job.enabled = true
spring.batch.initializer.enabled=true

4. mybatis 配置

import javax.sql.DataSource;

import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;

import com.alibaba.druid.pool.DruidDataSource;

import com.BlackListDao;
import com.ProperitesUtil;

/**
* mybatis配置
*
*/
@Configuration
public class MyBatisConfiguration {

private static final Logger logger = LoggerFactory.getLogger(MyBatisConfiguration.class);

@Autowired
SqlSessionFactory sqlSessionFactory;
@Autowired
SqlSessionTemplate sessionTemplate;

@Bean
public SqlSessionTemplate sqlSessionTemplate() {
return new SqlSessionTemplate(sqlSessionFactory);
}

@Bean
public BlackListDao blackListMapper() {
return sessionTemplate.getMapper(BlackListDao.class);
}

@Bean
@Primary
@ConfigurationProperties(prefix = "spring.datasource")
public DataSource dataSource() {
logger.debug("初始化dataSource");
return new DruidDataSource();
}

@Bean
public PlatformTransactionManager transactionManager(DataSource dataSource) {
PlatformTransactionManager transactionManager = new DataSourceTransactionManager(dataSource);
return transactionManager;
}

@Bean
public SqlSessionFactory sqlSessionFactory() throws Exception {

SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
sqlSessionFactoryBean.setDataSource(dataSource());

PathMatchingResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();

// 获取数据库类型
String dataBaseType = ProperitesUtil.getPropertyValue("spring.boot.database") == null ? "mysql"
: ProperitesUtil.getPropertyValue("spring.boot.database");

String directory = "classpath:/mapper/" + dataBaseType.trim().toLowerCase() + "/*.xml";
sqlSessionFactoryBean.setMapperLocations(resolver.getResources(directory));

return sqlSessionFactoryBean.getObject();
}

}

5. pom.xml文件

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion>

<groupId>com.spring.batch</groupId>
<artifactId>batch-test</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>

<name>batch-test</name>
<url>http://maven.apache.org</url>

<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.2.5.RELEASE</version>
</parent>

<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.batch</groupId>
<artifactId>spring-batch-core</artifactId>
<version>3.0.4.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-tomcat</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.tomcat.embed</groupId>
<artifactId>tomcat-embed-jasper</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>javax.servlet</groupId>
<artifactId>jstl</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>

</project>