tags：springbatch

1.引言

上一篇文章《快速使用組件-spring batch(3)讀文件數(shù)據(jù)到數(shù)據(jù)庫》對Spring Batch的讀碍庵、處理、寫組件進行了介紹绝编，并且以實際案例使用了FlatFileItemReader讀文本文件恃轩，并把每行數(shù)據(jù)映射為實體，然后使用JdbcBatchItemWriter把實體對象數(shù)據(jù)存儲到MySQL中锌畸。但在數(shù)據(jù)集成的實際應(yīng)用中勇劣，更多的工作是從A數(shù)據(jù)庫到B數(shù)據(jù)庫靖避，數(shù)據(jù)庫之間是異構(gòu)的，數(shù)據(jù)與數(shù)據(jù)的字段定義是不盡相同的比默，因此幻捏，在數(shù)據(jù)同步、數(shù)據(jù)抽取時命咐，需要從源數(shù)據(jù)庫讀取數(shù)據(jù)篡九，經(jīng)過校驗、轉(zhuǎn)換醋奠，過濾榛臼、清洗，然后把數(shù)據(jù)再寫到目標數(shù)據(jù)庫窜司。本文將在上一篇的基礎(chǔ)上沛善，實現(xiàn)數(shù)據(jù)庫到數(shù)據(jù)庫的數(shù)據(jù)同步。簡單來說塞祈，只需要把從文件讀數(shù)據(jù)改為從數(shù)據(jù)庫讀即可金刁。可下載完整示例工程代碼參考

數(shù)據(jù)庫到數(shù)據(jù)庫

2.開發(fā)環(huán)境

• JDK: jdk1.8
• Spring Boot: 2.1.4.RELEASE
• Spring Batch:4.1.2.RELEASE
• 開發(fā)IDE: IDEA
• 構(gòu)建工具Maven: 3.3.9
• 日志組件logback:1.2.3
• lombok:1.18.6

3.開發(fā)流程

上一篇文章中，已經(jīng)把User數(shù)據(jù)存儲在mytest數(shù)據(jù)庫中尤蛮，本文將以mytest數(shù)據(jù)庫中的test_user表為源數(shù)據(jù)媳友，使用Spring Batch把數(shù)據(jù)同步到目標數(shù)據(jù)庫my_test1，實現(xiàn)MySQL到MySQL的同步产捞。注醇锚，若是不同的數(shù)據(jù)庫，在配置多數(shù)據(jù)源時可更改數(shù)據(jù)庫驅(qū)動和連接信息即可坯临。關(guān)鍵代碼如下所示：

關(guān)鍵代碼

3.1 創(chuàng)建目標數(shù)據(jù)庫

在MySQL中創(chuàng)建my_test1數(shù)據(jù)庫作為目標數(shù)據(jù)庫搂抒，執(zhí)行示例工程中的sql/my_test1.sql創(chuàng)建用戶表，為簡單起見尿扯，目標數(shù)據(jù)表與源數(shù)據(jù)表數(shù)據(jù)表結(jié)構(gòu)一樣求晶。創(chuàng)建后如下：

目標數(shù)據(jù)庫

3.2 配置多數(shù)據(jù)源

至此，我們的程序涉及三個數(shù)據(jù)庫衷笋，分別是：

• 用于Spring Batch數(shù)據(jù)存儲的my_spring_batch為
• 源數(shù)據(jù)庫mytest
• 目標數(shù)據(jù)庫my_test1

因此芳杏，需要先配置多數(shù)據(jù)源，配置方法跟之前一樣辟宗，配置properties文件的數(shù)據(jù)庫連接信息和使用注解進行配置即可爵赵。如下：

application.properties

# spring batch db
spring.datasource.jdbc-url=jdbc:mysql://localhost:3310/my_spring_batch?serverTimezone=GMT%2B8&useUnicode=true&characterEncoding=utf8&useSSL=false
spring.datasource.username=root
spring.datasource.password=111111
# origin db
spring.origin-datasource.jdbc-url=jdbc:mysql://localhost:3310/mytest?serverTimezone=GMT%2B8&useUnicode=true&characterEncoding=utf8&useSSL=false
spring.origin-datasource.username=root
spring.origin-datasource.password=111111
# target db
spring.target-datasource.jdbc-url=jdbc:mysql://localhost:3310/my_test1?serverTimezone=GMT%2B8&useUnicode=true&characterEncoding=utf8&useSSL=false
spring.target-datasource.username=root
spring.target-datasource.password=111111

然后使用注解注入多數(shù)據(jù)源，如下：

DataSourceConfig.java

@Bean("datasource")
@ConfigurationProperties(prefix="spring.datasource")
@Primary
public DataSource batchDatasource() {
    return DataSourceBuilder.create().build();
}

@Bean("originDatasource")
@ConfigurationProperties(prefix="spring.origin-datasource")
public DataSource originDatasource() {
    return DataSourceBuilder.create().build();
}

@Bean("targetDatasource")
@ConfigurationProperties(prefix="spring.target-datasource")
public DataSource targetDatasource() {
    return DataSourceBuilder.create().build();
}

3.3 添加讀數(shù)據(jù)組件JdbcCursorItemReader

從數(shù)據(jù)庫中讀取數(shù)據(jù)泊脐，Spring Batch提供了組件JdbcCursorItemReader空幻，通過它，可以把數(shù)據(jù)庫的數(shù)據(jù)讀取出來容客，然后映射為實體User秕铛，以供后續(xù)開發(fā)。創(chuàng)建方法如下：

@Bean
public ItemReader db2DbItemReader(@Qualifier("originDatasource") DataSource originDatasource) {
    String readSql = " select * from test_user";
    return new JdbcCursorItemReaderBuilder<User>()
            .dataSource(originDatasource).sql(readSql)
            .verifyCursorPosition(false).rowMapper(new UserRowMapper())
            .build();
}

說明：

• 使用@Qualifier("originDatasource")標識源數(shù)據(jù)庫
• JdbcCursorItemReaderBuilder用于構(gòu)建JdbcCursorItemReader
• 讀數(shù)據(jù)的sql語句根據(jù)實際情況編寫即可缩挑，此處是讀取整個表數(shù)據(jù)但两。
• 需要把數(shù)據(jù)庫映射為實體User，使用UserRowMapper供置，此mapper實現(xiàn)RowMapper接口谨湘，把從數(shù)據(jù)庫讀取的ResultSet映射為User的字段。

3.4 自定義處理組件Db2DbItemProcessor

讀取到數(shù)據(jù)后芥丧，當前的處理是針對title字段紧阔，不為null的則轉(zhuǎn)為大寫即可。如下：

if(Objects.nonNull(title)){
    user.setTitle(title.toUpperCase());
}

3.5 添加寫數(shù)據(jù)組件JdbcBatchItemWriter

寫入數(shù)據(jù)续担，同樣使用JdbcBatchItemWriter組合擅耽，編寫插入sql語句，把實體User數(shù)據(jù)插入到數(shù)據(jù)庫即可赤拒，如下：

@Bean
public ItemWriter db2DbWriter(@Qualifier("targetDatasource") DataSource targetDatasource) {
    String inserSql ="INSERT INTO test_user(id,name,phone,title,email,gender,date_of_birth,sys_create_time,sys_create_user,sys_update_time,sys_update_user) " +
            "VALUES (:id,:name,:phone,:title,:email,:gender,:dateOfBirth,:sysCreateTime,:sysCreateUser,:sysUpdateTime,:sysUpdateUser)";
    return new JdbcBatchItemWriterBuilder<User>()
            .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
            .sql(inserSql)
            .dataSource(targetDatasource)
            .build();
}

說明：

• 使用JdbcBatchItemWriterBuilder進行JdbcBatchItemWriter的創(chuàng)建秫筏，設(shè)置插入數(shù)據(jù)庫的sql語句诱鞠，同時指定數(shù)據(jù)源即可。
• @Qualifier("targetDatasource") DataSource datasource用于指定數(shù)據(jù)源
• 使用BeanPropertyItemSqlParameterSourceProvider可以直接把讀取的數(shù)據(jù)實體的屬性數(shù)據(jù)作為參數(shù)填充到sql語句中这敬，從而實現(xiàn)數(shù)據(jù)插入操作航夺。

3.6 組裝完整任務(wù)

經(jīng)過上面的操作，可以使用一個java配置崔涂，把讀阳掐、寫、處理組裝成完整的step和job冷蚂，如下所示（詳細可見示例工程文件）：

@Bean
public Job db2DbJob(Step db2DbStep,JobExecutionListener db2DbListener){
    String funcName = Thread.currentThread().getStackTrace()[1].getMethodName();
    return jobBuilderFactory.get(funcName)
            .listener(db2DbListener)
            .flow(db2DbStep)
            .end().build();
}
@Bean
public Step db2DbStep(ItemReader db2DbItemReader ,ItemProcessor db2DbProcessor
        ,ItemWriter db2DbWriter){
    String funcName = Thread.currentThread().getStackTrace()[1].getMethodName();
    return stepBuilderFactory.get(funcName)
            .<User,User>chunk(10)
            .reader(db2DbItemReader)
            .processor(db2DbProcessor)
            .writer(db2DbWriter)
            .build();
}

3.7 測試

參考上一文章的File2DbJobTest缭保，編寫Db2DbJobTest文件即可。如下：

@Test
public void testDb2DbJob() throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
    //構(gòu)建job參數(shù)
    JobParameters jobParameters = JobUtil.makeJobParameters();
    //運行job
    Map<String, Object> stringObjectMap = jobLauncherService.startJob(db2DbJob, jobParameters);
    //測試結(jié)果
    Assert.assertEquals(ExitStatus.COMPLETED,stringObjectMap.get(SyncConstants.STR_RETURN_EXITSTATUS));
}

經(jīng)過此測試蝙茶，可查看到源數(shù)據(jù)庫mytest中的test_user表中的數(shù)據(jù)艺骂，已全部同步到目標庫my_test1中的test_user中。完成數(shù)據(jù)庫到數(shù)據(jù)庫的數(shù)據(jù)同步隆夯。

4.總結(jié)

本文通過簡單的示例钳恕，從源數(shù)據(jù)庫中讀取表數(shù)據(jù)，經(jīng)過處理蹄衷，寫入到目標數(shù)據(jù)庫忧额，具體一定的通用性。希望讓大家更深入的了解Spring Batch愧口，并能用到實踐中睦番。