这次写的功能是 实现springbatch的job每次调用、每次都会执行job里查询mysql列表的reader。因为碰到一个问题，批处理服务启动之后，调用job时，springbatch始终只处理一次，下次调用 显示已完成，原因就是初始化列表是服务启动时填充，只会填充一次，使用完之后不会再次填充列表，所以下次进入job数据为空。我们现在就解决这个问题，让job的每次调用都会从数据库获取最新的列表，并且批处理掉这份数据。

继上一次springbatch读取rabbitmq文章 继续编写。

因为采用的是通用设计，所以使用mybatis读取mysql列表同样也是。

废话不多说我们这次使用到的核心是 MyBatisCursorItemReader 。MyBatisCursorItemReader 是mybatis专门为了springbatch游标读取编写的。

首先我们修改原来的BatchConfig类 加入sqlSessionFactory方法，此处实例化使用的单例，因为使用到的次数比较多，为了节约内存资源，所以这么设计。

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;

import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.data.mongodb.core.MongoTemplate;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

@Configuration

public class BatchConfig {

@Autowired

protected JobBuilderFactory jobBuilderFactory;

@Autowired

protected StepBuilderFactory stepBuilderFactory;

@Autowired

protected RabbitTemplate amqpTemplate;

@Autowired

protected MongoTemplate mongoTemplate;

@Autowired

private MybatisProperties properties;

private static SqlSessionFactoryBean sessionFactory;

//构造线程

@Bean

protected ThreadPoolTaskExecutor taskExecutor(){

ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

executor.setCorePoolSize(10);

executor.setMaxPoolSize(15);

executor.setKeepAliveSeconds(300);

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());

executor.setQueueCapacity(10000);

executor.setThreadGroupName("spring_batch");

return executor;

}

public SqlSessionFactory sqlSessionFactory() {

try {

if(sessionFactory == null){

sessionFactory = new SqlSessionFactoryBean();

sessionFactory.setDataSource(SpringContextHolder.getBean("dataSource"));

sessionFactory.setMapperLocations(properties.resolveMapperLocations());

return sessionFactory.getObject();

}

return sessionFactory.getObject();

} catch (Exception e) {

e.printStackTrace();

return null;

}

}

}

另外编写了专门使用mybatis循环读取mysql列表

import org.apache.ibatis.session.SqlSessionFactory;

import org.mybatis.spring.batch.MyBatisCursorItemReader;

import org.springframework.batch.item.ItemReader;

import java.util.Map;

/**

* Description: 用mybatis返回list 循环读取值

* @date: 2019/4/28 9:31

*/

public class ReadToListByMybatis {

/**

* Description: 读取mybatis list

* @param: 查询方法需要的参数 ，queryId 可反射的mybatis接口的ID

* example：

* parameterValues.put("createDate", "2019-04-28");

* parameterValues.put("queryId", "com.xx.xx.batch.dao.XXDao.select");

*

* @date: 2019/4/28 9:28

*/

public static ItemReader myBatisCursorItemReader(MapparameterValues, SqlSessionFactory sqlSessionFactory) {

MyBatisCursorItemReader reader = new MyBatisCursorItemReader();

reader.setSqlSessionFactory(sqlSessionFactory);

reader.setParameterValues(parameterValues);

reader.setQueryId(parameterValues.get("queryId").toString());

return reader;

}

}

现在我们修改自定义config 使用mybatis从mysql中读取

import com.xx.xx.config.BatchConfig;

import com.xx.xx.batch.dao.UserDao;

import com.xx.xx.batch.entity.User;

import com.xx.xx.batch.listener.UserJobCompletionListener;

import com.xx.xx.batch.processor.UserProcessor;

import com.xx.xx.batch.read.RabbitRead;

import com.xx.xx.batch.writer.MongoWriter;

import com.xx.xx.batch.writer.MysqlWriter;

import org.springframework.batch.core.Job;

import org.springframework.batch.core.JobExecutionListener;

import org.springframework.batch.core.Step;

import org.springframework.batch.core.launch.support.RunIdIncrementer;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

@Configuration

public class UserBatchConfig extends BatchConfig {

@Resource

private UserDao userDao;

private Mapparam;

@Bean

public Step userStep(){

param.put("queryId", "com.xx.xx.batch.dao.UserDao.select");

return stepBuilderFactory.get("userStep").chunk(10000)

.reader(ReadToListByMybatis.myBatisCursorItemReader(param,sqlSessionFactory())).processor(new UserProcessor()).writer(new MysqlWriter(userDao)).taskExecutor(taskExecutor())

.build();

}

@Bean

public Step userStep2(){

//业务逻辑可自写

/*param.put("queryId", "com.xx.xx.batch.dao.UserDao.select");

return stepBuilderFactory.get("userStep").chunk(10000)

.reader(ReadToListByMybatis.myBatisCursorItemReader(param,sqlSessionFactory())).processor(new UserProcessor()).writer(new MysqlWriter(userDao)).taskExecutor(taskExecutor())

.build();*/

}

@Bean

public Job processJob1(){

param = new HashMap();

param.put("createDate", "2019-04-28");

// 此处userStep2 代表可以使用多个step 执行完第一个step之后就开始执行step2

return jobBuilderFactory.get("processJob1").

incrementer(new RunIdIncrementer()).listener(listener()).

flow(userStep()).next(userStep2()).end().build();

}

// 监听事件

@Bean

public JobExecutionListener listener() {

return new UserJobCompletionListener();

}

}

接下来我们修改UserDao增加id方法

@Mapper

public interface UserDaoextends MysqlCommonDao{

@Override

Integer add(T user);

@Override

List select(Mapmap);

}

在修改mapper文件

insert into user_batch_test values(#{id},#{age},#{name})

SELECT * FROM user

1=1 and create_date IS NOT NULL

and create_date = #{date}

最后 ，在启动项目的时候 springbatch会初始化任务以及任务里的reader，在调用任务job时，批处理掉从mysql查出来的列表。

下次调用任务job时，又会重新的读取mysql查询最新列表并处理。