tags：springbatch

1.引言

上一篇《便捷的数据读写-spring batch(5)结合beetlSql进行数据读写》中使用Spring Batch及BeetlSql，对数据库读写组件进行数据库同步，实际上是全量同步。全量同步的问题在于每次需要读取整个表数据，如果表数据量大，则资源耗费大，而且不便于对已有数据的更新。因此，在数据同步过程中，更多的使用增量同步，即通过某些条件，区分新数据进行插入，对有变化的数据进行更新，对不存在的数据进行删除等(当然，一般都不会对数据进行物理删除，只做逻辑删除，因此也就变成了数据更新操作)。

增量更新更多情况需要依据上一次更新后的状态(如时间、自增ID，数据位置等)，下一次更新以上一次更新的状态为基础，因此，需要把每一次更新后的状态以变量参数的方式保存下来，下一次更新则以此状态数据为动态参数来使用。Spring Batch支持任务运行时的动态参数，结合此特性，可以实现数据的增量同步。

2.开发环境

JDK: jdk1.8

Spring Boot: 2.1.4.RELEASE

Spring Batch:4.1.2.RELEASE

开发IDE: IDEA

构建工具Maven: 3.3.9

日志组件logback:1.2.3

lombok:1.18.6

3.增量同步简述

增量同步，是相对与全量同步，即每次同步，只需要同步源数据库变化的部分，这样提高了数据同步的效率。是当前数据同步的普遍方式。抽取变化的数据，又名CDC，即Change Data Capture变化数据捕获。在《Pentaho Kettle解决方案：使用PDI构建开源ETL解决方案 》一书中，对CDC作了比较详细说明。此处简要做一下说明，当前实现增量同步的方式有4种，分别是基于源数据的CDC，基于触发器的CDC，基于快照的CDC，基于日志的CDC。

3.1 基于源数据的CDC

基于源数据的CDC要求源数据里有相关的属性列，利用这些属性列，可以判断出哪里是增量数据，最常见的属性列有：

时间戳

基于时间来标识数据，至少需要一个时间，最好两个，一个标识创建，一个标识更新时间，所以一般我们设计数据库时都会添加sys_create_time和sys_update_time作为默认字段，并且设计为默认当前时间和更新处理。

自增序列

使用数据库表的自增序列字段(一般是主键)，来标识新插入的数据。不过现实中用得比较少。

此方法需要有一个临时表来保存上一次更新时间或，在实践中，一般是在独立的模式下创建此表，保存数据。下一次更新则比较上一次时间或序列。这是用得比较普遍的方式，本文中的增量同步也是使用此方法。

3.2 基于触发器的CDC

在数据库中编写触发器，当前数据库执行INSERT，UPDATE，DELETE等语句时，可以激活数据库中的触发器，然后触发器可以把这些变更的数据保存到中间临时表，然后再从临时表中获取这些数据，同步到目标数据库中。当然，这种方法是入侵性最强的，一般数据库都不允许向数据库里添加触发器(影响性能)。

3.3 基于快照的CDC

此方法就是一次抽取当前全部数据放到缓冲区，作为快照，下一次同步时从源数据读取数据，然后和快照做比较，找出变化的数据。简单来说是就做全表读取与比较，找出变化的数据。做全表扫描，问题就在于性能，所以一般不会使用这种方式。

3.4 基于日志的CDC

最高级和最没有入侵性的方法就是基于日志的方式，数据库会把插入、更新、删除的操作记到日志中，如Mysql会有binlog，增量同步可以读取日志文件，把二进制文件转为可理解的方式，然后再把里面的操作按照顺序重做一遍。但是这种方式只能对同种数据库有效，对于异构的数据库就无法实现了。而且实现起来有一定的难度。

3.5 本示例增量同步方法说明

在本示例中，依然是基于test_user表进行增量同步，表有字段sys_create_time和sys_update_time来标识数据创建和更新时间(当前，若现实情况中只有一个时间，也可以只基于此时间，只是这样就比较难标识此数据是更新还是插入了)。增量同步流程如下：

流程

说明：

每次同步，会先读取临时表，获取上一次同步后数据的时间。

若是第一次同步，则全部同步，若不是，则根据时间作为查询语句的参数。

根据时间读取数据后，把数据插入目标表

更新临时表的数据时间，以便下一次同步。

4.Spring Batch动态参数绑定

根据上面的增量同步流程，关键点在于把数据时间保存到临时表，在数据读取时可以作为比较的条件。而这时间参数是动态的，在任务执行时才传递进去，在Spring Batch中，支持动态参数绑定，只需要使用@StepScope注解即可，结合BeetlSql，很快就可以实现增量同步。本示例是基于上一篇文章的示例来进一步开发的，可以下载源码查看完整示例。

4.1 沿用原来数据库配置和多数据源

源数据库: mytest

目标数据库: my_test1

spring batch数据库: my_spring_batch

同步的数据表：test_user

4.2 创建临时表

使用示例中的sql/initCdcTempTable.sql，在my_spring_batch库中，创建临时表cdc_temp，并插入记录为1的记录，标识是同步test_user表。此处，我们只需要关注last_update_time和current_update_time，前者表示上一次同步完后的数据最后时间，后者表示上一次同步后的系统时间。

4.3 添加/修改dao

4.3.1 添加临时表dao及service类

添加类CdcTempRepository

根据配置，由于cdc_temp是在my_spring_batch，而它的读写是在dao.local包中，因此需要添加dao.local包，然后添加类CdcTempRepository，如下所示：

@Repository

public interface CdcTempRepository extends BaseMapper {

}

添加类CdcTempService，用于cdc_temp表的读取及数据更新

主要包括两个函数，一个是根据ID获取当前的cdc_temp记录，以便获取数据上一次同步的数据最后时间。一个是在同步完成后，更新cdc_temp的数据。如下：

/**

* 根据id获取cdc_temp的记录

* @param id 记录ID

* @return {@link CdcTemp}

*/

public CdcTemp getCurrentCdcTemp(int id){

return cdcTempRepository.getSQLManager().single(CdcTemp.class, id);

}

/**

* 根据参数更新cdcTemp表的数据

* @param cdcTempId cdcTempId

* @param status job状态

* @param lastUpdateTime 最后更新时间

*/

public void updateCdcTempAfterJob(int cdcTempId,BatchStatus status,Date lastUpdateTime){

//获取

CdcTemp cdcTemp = cdcTempRepository.getSQLManager().single(CdcTemp.class, cdcTempId);

cdcTemp.setCurrentUpdateTime(DateUtil.date());

//正常完成则更新数据时间

if( status == BatchStatus.COMPLETED){

cdcTemp.setLastUpdateTime(lastUpdateTime);

}else{

log.info(LogConstants.LOG_TAG+"同步状态异常："+ status.toString());

}

//设置同步状态

cdcTemp.setStatus(status.name());

cdcTempRepository.updateById(cdcTemp);

}

4.3.2 修改源数据dao

在源数据dao类OriginUserRepository添加函数getOriginIncreUser，此函数对应user.md中的sql语句。

4.3.3 修改目标数据dao

在目标数据dao类TargetUserRepository中添加函数selectMaxUpdateTime，用于查询同步后数据的最后时间。由于此方法的sql简单，可以直接使用@Sql注解，如下所示：

@Sql(value="select max(sys_update_time) from test_user")

Date selectMaxUpdateTime();

4.4 修改user.md中的sql语句。

4.4.1 添加增量读数据sql

在user.md中添加增量读数据的sql语句，如下：

getOriginIncreUser

===

* 查询user数据

select * from test_user

WHERE 1=1

@if(!isEmpty(lastUpdateTime)){

AND (sys_create_time >= #lastUpdateTime# OR sys_update_time >= #lastUpdateTime#)

@}

说明：

@开头是beetl的语法，可以对变量读取和逻辑判断，此处的意思是如果变量lastUpdateTime不为空，则按此条件进行读取。

lastUpdateTime变量由调用时传入(Map)

具体beetl使用语法，可参见官方文档

4.4.2 编写增量插入sql语句

对于Mysql数据库，有insert into ... on duplicate key update ...的用法，即可以根据唯一键(主键或唯一索引)，若数据已存在，则更新，不存在，则插入。在user.md文件中，添加以下语句：

insertIncreUser

===

* 插入数据

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#)

ON DUPLICATE KEY UPDATE

id = VALUES(id),

name = VALUES(name),

phone = VALUES(phone),

title = VALUES(title),

email = VALUES(email),

gender = VALUES(gender),

date_of_birth = VALUES(date_of_birth),

sys_create_time = VALUES(sys_create_time),

sys_create_user = VALUES(sys_create_user),

sys_update_time = VALUES(sys_update_time),

sys_update_user = VALUES(sys_update_user)

4.5 编写Spring Batch的组件

Spring Batch文件结构如下：

文件结构

4.5.1 ItemReader

此处与之前的一致，只需要把getOriginUser函数改为getOriginIncreUser即可。

4.5.2 ItemWriter

此处与之前的一致，只需要把sql的ID由user.insertUser改为user.insertIncreUser即可。

4.5.3 添加IncrementJobEndListener

由于数据同步完后，最后一步就是要更新临时表的最后时间数据。如下：

@Slf4j

public class IncrementJobEndListener extends JobExecutionListenerSupport {

@Autowired

private CdcTempService cdcTempService;

@Autowired

private TargetUserRepository targetUserRepository;

@Override

public void afterJob(JobExecution jobExecution) {

BatchStatus status = jobExecution.getStatus();

Date latestDate = targetUserRepository.selectMaxUpdateTime();

cdcTempService.updateCdcTempAfterJob(SyncConstants.CDC_TEMP_ID_USER,status,latestDate);

}

}

说明：

先查询当前数据库中数据最后时间(selectMaxUpdateTime)

更新中间表数据cdc_temp中的last_update_time

4.5.4 添加任务启动时参数初始化

在数据同步的第一步，需要先初始化临时表中的数据最后更新时间，因此在任务启动前，先要进行任务参数设置，以便于把时间参数传到任务中，在任务执行时使用。如下：

public JobParameters initJobParam(){

CdcTemp currentCdcTemp = cdcTempService.getCurrentCdcTemp(getCdcTempId());

//若未初始化，则先查询数据库中对应的最后时间

if(SyncConstants.STR_STATUS_INIT.equals(currentCdcTemp.getStatus())

|| SyncConstants.STR_STATUS_FAILED.equals(currentCdcTemp.getStatus())){

Date maxUpdateTime = selectMaxUpdateTime();

//若没有数据，则按初始时间处理

if(Objects.nonNull(maxUpdateTime)){

currentCdcTemp.setLastUpdateTime(maxUpdateTime);

}

}

return JobUtil.makeJobParameters(currentCdcTemp);

}

4.5.5 组装完整任务

最后，需要一个IncrementBatchConfig配置把读、处理、写、监听组装起来，值得一提的是，在配置读组件时，由于需要使用动态参数，此处需要添加@StepScope注解，同时在参数中使用spEL获取参数内容，如下所示：

@Bean

@StepScope

public ItemReader incrementItemReader(@Value("#{jobParameters['lastUpdateTime']}") String lastUpdateTime) {

IncrementUserItemReader userItemReader = new IncrementUserItemReader();

//设置参数，当前示例可不设置参数

Map params = CollUtil.newHashMap();

params.put(SyncConstants.STR_LAST_UPDATE_TIME,lastUpdateTime);

userItemReader.setParams(params);

return userItemReader;

}

4.5.6 测试

参考上一文章的BeetlsqlJobTest，编写IncrementJobTest测试文件。由于需要测试增量同步，测试流程如下所示：

测试前增量添加数据

测试前，源数据表和目标数据表已经有数据，在源数据表中，执行代码中的sql/user-data-new.sql添加新的用户。注意，由于sys_create_time和sys_update_time定义如下：

`sys_create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,

`sys_update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

从而达到数据插入时自动生成时间，修改时也自动更新时间。

运行测试

以单元测试运行incrementJob。

查看结果

运行完成后，结果如下：

输出

增量同步后，数据如下：

结果

5.总结

本文先对增量同步做了一个简单的介绍，列出当前一般使用的增量同步方法，然后使用Spring Batch和BeetlSql使用基于时间戳的方式实现增量同步，本示例具有一定的实用性，希望能对做数据同步或相关批处理的开发者有帮助。