我将 Spring Batch 与 MyBatis 结合使用，使用 MyBatisBatchItemWriter 将文件数据写入 MySql 数据库。
以下是配置和代码实现:
批量配置:
@Autowired
private SqlSessionFactory sqlSessionFactory;
@Bean
public Job userPostJob() throws IOException {
log.info("Begin userPostJob Job");
return jobBuilderFactory.get("userPostJob")
.incrementer(new RunIdIncrementer()).listener(userPostJobCompletionNotificationListener)
.flow(userPostFlow())
.end()
.build();
}
private Step userPostFlow() {
return stepBuilderFactory.get("userWriterFlow").<UserTable, UserTable>chunk(2)
.reader(getUserReader(null))
.processor(userProcessor)
.writer(myBatisUserBatchWriter())
.build();
}
@Bean
@StepScope
public FlatFileItemReader<UserTable> getUserReader(
@Qualifier("userFileTokenizer") LineTokenizer userFileTokenizer) {
return reader("/Users/aditya.singh/user.csv", true,
sagawaFileTokenizer, UserTable.class);
}
private <T> FlatFileItemReader<T> reader(String filePath, boolean skipHeaderLine,
LineTokenizer lineTokenizer, Class<T> clazz) {
FlatFileItemReader<T> flatFileItemReader = new FlatFileItemReader<>();
flatFileItemReader.setResource(new FileSystemResource(filePath));
flatFileItemReader.setLinesToSkip(skipHeaderLine ? 1 : 0);
DefaultLineMapper<T> defaultLineMapper = new DefaultLineMapper<>();
defaultLineMapper.setLineTokenizer(lineTokenizer);
defaultLineMapper.setFieldSetMapper(createFieldMapper(clazz));
flatFileItemReader.setLineMapper(defaultLineMapper);
return flatFileItemReader;
}
private <T> FieldSetMapper<T> createFieldMapper(Class<T> clazz) {
BeanWrapperFieldSetMapper<T> beanWrapperFieldSetMapper = new BeanWrapperFieldSetMapper<>();
beanWrapperFieldSetMapper.setTargetType(clazz);
return beanWrapperFieldSetMapper;
}
public MyBatisBatchItemWriter<User> myBatisUserBatchWriter() {
MyBatisBatchItemWriter<UserTable> cvsInfoWriter = new MyBatisBatchItemWriter();
cvsInfoWriter.setSqlSessionFactory(sqlSessionFactory);
cvsInfoWriter
.setStatementId("com.batch.mapper.UserTableMapper.insert");
return cvsInfoWriter;
}
数据源配置:
@Configuration
@ConfigurationProperties(prefix = "database")
@Setter
@EnableTransactionManagement
public class DataSourceConfiguration {
private String driverClass;
private String databaseUrl;
private String username;
private String password;
private Integer maxPoolSize;
@Bean
public DataSource dataSource() {
HikariConfig config = new HikariConfig();
config.setDriverClassName(driverClass);
config.setJdbcUrl(databaseUrl);
config.setUsername(username);
config.setPassword(password);
config.setMaximumPoolSize(maxPoolSize);
return new HikariDataSource(config);
}
@Bean
public SqlSessionFactory sqlSessionFactory() throws Exception {
SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
sqlSessionFactoryBean.setDataSource(dataSource());
return sqlSessionFactoryBean.getObject();
}
@Bean
public DataSourceTransactionManager dataSourceTransactionManager() {
DataSourceTransactionManager dataSourceTransactionManager = new DataSourceTransactionManager();
dataSourceTransactionManager.setDataSource(dataSource());
return dataSourceTransactionManager;
}
}
MyBatis 配置和生成的 Mapper 接口(interface):
generatorConfig.xml
<commentGenerator> <property name="suppressDate" value="true"/> </commentGenerator> <jdbcConnection driverClass="com.mysql.cj.jdbc.Driver" connectionURL="jdbc:mysql://localhost:3306/employee" userId="root" password="root"> </jdbcConnection> <javaTypeResolver> <property name="forceBigDecimals" value="false"/> </javaTypeResolver> <javaModelGenerator targetPackage="com.batch.model" targetProject="src/main/java"> <property name="enablesubpackages" value="true"/> <property name="trimstrings" value="true"/> </javaModelGenerator> <javaClientGenerator type="XMLMAPPER" targetPackage="com.batch.mapper" targetProject="src/main/java"> <property name="enableSubPackages" value="true"/> </javaClientGenerator> <table tableName="USER_TABLE"> <property name="useActualColumnNames" value="false"/> </table>生成的Mapper类:
@ToString
public class UserTable {
@Generated(value = "org.mybatis.generator.api.MyBatisGenerator", comments = "Source field: USER_TABLE.id")
private String id;
@Generated(value = "org.mybatis.generator.api.MyBatisGenerator", comments = "Source field: USER_TABLE.name")
private String name;
@Generated(value = "org.mybatis.generator.api.MyBatisGenerator", comments = "Source field: USER_TABLE.id")
public String getId() {
return id;
}
@Generated(value = "org.mybatis.generator.api.MyBatisGenerator", comments = "Source field: USER_TABLE.id")
public void setId(String id) {
this.id = id;
}
@Generated(value = "org.mybatis.generator.api.MyBatisGenerator", comments = "Source field: USER_TABLE.name")
public String getName() {
return name;
}
@Generated(value = "org.mybatis.generator.api.MyBatisGenerator", comments = "Source field: USER_TABLE.name")
public void setName(String name) {
this.name = name;
}
}
@Mapper
public interface UserTableMapper {
@Generated(value="org.mybatis.generator.api.MyBatisGenerator", comments="Source Table: USER_TABLE")
@InsertProvider(type=SqlProviderAdapter.class, method="insert")
int insert(InsertStatementProvider<UserTable> insertStatement);
@Generated(value="org.mybatis.generator.api.MyBatisGenerator", comments="Source Table: USER_TABLE")
default int insert(UserTable record) {
return insert(SqlBuilder.insert(record)
.into(userTable)
.map(id).toProperty("id")
.map(name).toProperty("name")
.build()
.render(RenderingStrategy.MYBATIS3));
}
@Generated(value="org.mybatis.generator.api.MyBatisGenerator", comments="Source Table: USER_TABLE")
default int insertSelective(UserTable record) {
return insert(SqlBuilder.insert(record)
.into(userTable)
.map(id).toPropertyWhenPresent("id", record::getId)
.map(name).toPropertyWhenPresent("name", record::getName)
.build()
.render(RenderingStrategy.MYBATIS3));
}
// Other update & select methods
}
public final class UserTableDynamicSqlSupport {
@Generated(value="org.mybatis.generator.api.MyBatisGenerator", comments="Source Table: USER_TABLE)
public static final UserTable userTable = new UserTable();
@Generated(value="org.mybatis.generator.api.MyBatisGenerator", comments="Source field: USER_TABLE.id")
public static final SqlColumn<String> id = userTable.id;
@Generated(value="org.mybatis.generator.api.MyBatisGenerator", comments="Source field: USER_TABLE.name")
public static final SqlColumn<String> name = userTable.name;
@Generated(value="org.mybatis.generator.api.MyBatisGenerator", comments="Source Table: USER_TABLE")
public static final class UserTable extends SqlTable {
public final SqlColumn<String> id = column("id", JDBCType.VARCHAR);
public final SqlColumn<String> name = column("name", JDBCType.VARCHAR);
public UserTable() {
super("USER_TABLE");
}
}
}
异常堆栈跟踪:
org.mybatis.spring.MyBatisSystemException: nested exception is org.apache.ibatis.builder.BuilderException: Error invoking SqlProvider method 'public java.lang.String org.mybatis.dynamic.sql.util.SqlProviderAdapter.insert(org.mybatis.dynamic.sql.insert.render.InsertStatementProvider)' with specify parameter 'class com.batch.model.UserTable'. Cause: java.lang.IllegalArgumentException: argument type mismatch
at org.mybatis.spring.MyBatisExceptionTranslator.translateExceptionIfPossible(MyBatisExceptionTranslator.java:78)
at org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:440)
at com.sun.proxy.$Proxy65.update(Unknown Source)
at org.mybatis.spring.SqlSessionTemplate.update(SqlSessionTemplate.java:287)
at org.mybatis.spring.batch.MyBatisBatchItemWriter.write(MyBatisBatchItemWriter.java:144)
at org.springframework.batch.core.step.item.SimpleChunkProcessor.writeItems(SimpleChunkProcessor.java:188)
at org.springframework.batch.core.step.item.SimpleChunkProcessor.doWrite(SimpleChunkProcessor.java:154)
at org.springframework.batch.core.step.item.SimpleChunkProcessor.write(SimpleChunkProcessor.java:287)
at org.springframework.batch.core.step.item.SimpleChunkProcessor.process(SimpleChunkProcessor.java:212)
at org.springframework.batch.core.step.item.ChunkOrientedTasklet.execute(ChunkOrientedTasklet.java:75)
at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:407)
at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:331)
at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:140)
at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:273)
at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:82)
at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:375)
at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215)
at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:145)
at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:258)
at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:203)
at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:148)
at org.springframework.batch.core.job.flow.JobFlowExecutor.executeStep(JobFlowExecutor.java:68)
at org.springframework.batch.core.job.flow.support.state.StepState.handle(StepState.java:67)
at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:169)
at org.springframework.batch.core.job.flow.support.SimpleFlow.start(SimpleFlow.java:144)
at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:136)
at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:313)
at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:144)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:137)
at com.batch.controller.JobLauncherController.launchUserPostJob(JobLauncherController.java:40)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:567)
at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:190)
at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:138)
at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:105)
at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:893)
at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:798)
at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)
at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1040)
at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:943)
at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1006)
at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:898)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:634)
at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:883)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:741)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:231)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.springframework.web.filter.HiddenHttpMethodFilter.doFilterInternal(HiddenHttpMethodFilter.java:94)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:201)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:202)
at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:96)
at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:526)
at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:139)
at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:92)
at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:74)
at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:343)
at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:408)
at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:66)
at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:860)
at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1589)
at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
at java.base/java.lang.Thread.run(Thread.java:835)
Caused by: org.apache.ibatis.builder.BuilderException: Error invoking SqlProvider method 'public java.lang.String org.mybatis.dynamic.sql.util.SqlProviderAdapter.insert(org.mybatis.dynamic.sql.insert.render.InsertStatementProvider)' with specify parameter 'class com.batch.model.UserTable'. Cause: java.lang.IllegalArgumentException: argument type mismatch
at org.apache.ibatis.builder.annotation.ProviderSqlSource.createSqlSource(ProviderSqlSource.java:151)
at org.apache.ibatis.builder.annotation.ProviderSqlSource.getBoundSql(ProviderSqlSource.java:113)
at org.apache.ibatis.mapping.MappedStatement.getBoundSql(MappedStatement.java:297)
at org.apache.ibatis.executor.statement.BaseStatementHandler.<init>(BaseStatementHandler.java:64)
at org.apache.ibatis.executor.statement.PreparedStatementHandler.<init>(PreparedStatementHandler.java:41)
at org.apache.ibatis.executor.statement.RoutingStatementHandler.<init>(RoutingStatementHandler.java:46)
at org.apache.ibatis.session.Configuration.newStatementHandler(Configuration.java:592)
at org.apache.ibatis.executor.BatchExecutor.doUpdate(BatchExecutor.java:57)
at org.apache.ibatis.executor.BaseExecutor.update(BaseExecutor.java:117)
at org.apache.ibatis.executor.CachingExecutor.update(CachingExecutor.java:76)
at org.apache.ibatis.session.defaults.DefaultSqlSession.update(DefaultSqlSession.java:197)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:567)
at org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:426)
... 83 common frames omitted
Caused by: java.lang.IllegalArgumentException: argument type mismatch
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:567)
at org.apache.ibatis.builder.annotation.ProviderSqlSource.invokeProviderMethod(ProviderSqlSource.java:191)
at org.apache.ibatis.builder.annotation.ProviderSqlSource.createSqlSource(ProviderSqlSource.java:134)
... 98 common frames omitted
我一直在调试这个问题，直到Java调用方法，但未能解决。
此外，如果我使用编写器类(如下)，它可以正常工作并保存数据。
@Component
public class UserWriter implements ItemWriter<UserTable> {
@Autowired
private UserTableMapper userTableMapper;
@Override
public void write(List<? extends UserTable> items) throws Exception {
for (UserTable userTable : items) {
log.info("Writing user info: {}", userTable);
userTableMapper.insert(userTable);
}
}
}
版本 (pom.xml):
// Spring Boot
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.9.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<java.version>1.8</java.version>
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>2.1.0</version>
</dependency>
<dependency>
<groupId>org.mybatis.dynamic-sql</groupId>
<artifactId>mybatis-dynamic-sql</artifactId>
<version>1.1.3</version>
</dependency>
<dependency>
<groupId>org.mybatis.generator</groupId>
<artifactId>mybatis-generator-maven-plugin</artifactId>
<version>1.3.7</version>
</dependency>
MyBatis 文档对此的信息非常有限，我觉得我已经按照 MyBatisBatchItemWriter 期望的方式完成了。
有人可以帮我找到这段小而简单的代码中的问题吗？
最佳答案
在当前实现中，MyBatisBatchItemWriter不支持调用mapper接口(interface)上的默认方法，因为它通过SqlSession调用mapper语句界面。在这种情况下，MyBatisBatchItemWriter调用int insert(InsertStatementProvider<UserTable> insertStatement);对应的mapper语句而不是int insert(UserTable record) 。因此，如果您想调用默认方法，您应该创建一个自定义项目编写器。作为替代方案，您可以使用 itemToParameterConverter用于转换为 InsertStatementProvider<UserTable> 的属性来自UserTable如下:
public MyBatisBatchItemWriter<UserTable> myBatisUserBatchWriter() {
MyBatisBatchItemWriter<UserTable> cvsInfoWriter = new MyBatisBatchItemWriter<>();
cvsInfoWriter.setSqlSessionFactory(sqlSessionFactory);
cvsInfoWriter.setStatementId("com.batch.mapper.UserTableMapper.insert");
cvsInfoWriter.setItemToParameterConverter(record -> SqlBuilder.insert(record)
.into(userTable)
.map(id).toProperty("id")
.map(name).toProperty("name")
.build()
.render(RenderingStrategy.MYBATIS3));
return cvsInfoWriter;
}
关于java - Spring Batch 和我的 Batis : MyBatisBatchItemWriter throws IllegalArgumentException: argument type mismatch，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/58334625/