基于Quartz的高频率同步

quartz是一个按照设定的时间规则来调度作业的调度器，比如可以设定每30s启动一个Job，但如果Job在30s内还未完成，那么quartz默认情况下还是按照设定的周期启动新的Job线程。

有这样一个应用场景，两个表结构相同的数据库A和B，A是一个实时交易的业务系统数据库，需要把A中的数据完全准实时的复制到B中。同时，交易表中存在一个时间戳字段，同步任务是每次按照这个时间戳来作增量同步的，即先在B中select max(col_timestamp)得到当前最大时间戳，再到A中获取大于该时间戳的数据集，最后再insert到B中。

在该场景下，如果前一个线程还未结束写的动作（比如一共1000条记录，已经提交了500条，还剩500条没提交），而后一个现成又开始读的话，将导致后500条记录重复插入的异常。实现的代码是完全的JDBC代码，不支持跨数据库级别的事务管理，而Job中第一次读和后面的写是在数据库B中，可以合并为一个事务(t1)，第二次读是针对数据库A，作为一个事务(t2)。为了避免以上的幻读和重复写异常，可以这样作：
(1)将connection的自动提交设为false。
(2)设置事务隔离级别为serialization。事务t1肯定要设置为最高级别serialization的，因为要避免幻读（避免当前一个线程还在写的时候，后续的线程来读）;而事务t2可以不用把隔离级别设为serialization，默认的read commited即可。
(3)每次把当前需要insert的数据作为一次性提交，也就是说第一次的读和后面的写作为一个事务而不是多个。

这样即时数据量再大，而同步周期再短，数据都能顺利的复制到目标库。

回过头来看整个流程，其实在数据库级别加serialization隔离级别，就是使得事务能够串行处理。理论上来说，这种效果其实在应用程序层面也是可以做到的，即后面的线程在前面的线程处理完成之后再启动，也就是所谓的状态化线程控制，但quartz似乎并不支持这一点。

评论

12 楼 dayang2001911 2008-11-04  

用quartz的statefuljob这个接口是么？
我想问问用这个可以保证不并发吗？怎么我测试这个JOB并且设定按照每分钟触发一次，不会阻塞并发运行呢？
public class AAAAJob implements StatefulJob {


/**
* @param context .
* @throws JobExecutionException .
*/
public void execute(JobExecutionContext context)
throws JobExecutionException {
try {
int count=0;
while(true){
log.info("count:"+(count++));
}
} catch (Exception ex) {
ex.printStackTrace();

}
}
}

11 楼 duobin3000 2008-06-27  

毛
瞎说，
qutarz有这样的实现，而且spring-qutarz也提供了这样的功能：
<property name="concurrent" value="false" />

10 楼 e-ant 2008-02-18  

我解决这个问题，是用了一个全局变量，来注释上一个job是否提交完成
每个job定时调度时，先判断上个job是否提交完成，完成则进行数据库操作，否则终止job

9 楼 simbasun 2008-02-15  

StatefulJob 是正解...

StatefulJob instances follow slightly different rules from regular Job instances. The key difference is that their associated JobDataMap is re-persisted after every execution of the job, thus preserving state for the next execution. The other difference is that stateful jobs are not allowed to execute concurrently, which means new triggers that occur before the completion of the execute(xx) method will be delayed.

8 楼 ddandyy 2008-02-15  

所以我说的是你那个是病句.........
首先你就给quartz定义为"按照设定的时间规则来调度作业的调度器" 而他实际上不只是这个, 其次你后面说的是"每30s启动一个Job"这个很明显是指的循环调用方式 而不是定时方式

P.S: 这个没什么好说的吧  在数据库加一个状态控制就实事了 第2个启动 发现状态还是启动中 就中止

7 楼 tigers 2008-02-15  

ddandyy 写道

引用

quartz是一个按照设定的时间规则来调度作业的调度器，比如可以设定每30s启动一个Job，但如果Job在30s内还未完成，那么quartz默认情况下还是按照设定的周期启动新的Job线程。

这个应该是一个病句
就我说知quartz的两种调用方式 一种定时运行 一种循环运行
定时运行不考虑前一个是否执行完毕  循环运行当前一个没执行完的情况下  是不会启动下一个job的

两年前专门test过这个东西......现在应该不会变吧.....这种基础的东西

对，是两种方式，但现在说的是定时方式，你说的手工循环方式没错，但我们不是在讨论它

6 楼 tigers 2008-02-15  

抛出异常的爱 写道

事务的颗度太大是你的问题根源....
你把所有的1000条数据都读出来了...
如果你不是一次读出所有的数据
那么你的问题是很简单的.
当第二个quartz起来时把前一个业务结束.

问题是事务的粒度不是同步本身能够控制的，而是由业务需求确定的。从系统的健壮性来看，即使数据粒度再小，也不能保证在网络和机器处理能力等因素影响下能在设定的固定周期内完成事务，所以还是得从系统本身考虑，无论从quartz的线程控制级别还是jdbc事务级别。

5 楼 tigers 2008-02-15  

statefuljob确实可以考虑，从实验来看，它是等待上一个周期的运行实例结束之后再启动下一个周期的实例，但问题如果实例运行持续的时间很长，比如是周期的10倍，那么quartz启动的是当前时间的实例还是补上这10个周期内没启动的实例呢？

从实验结果来看，我把周期设为1秒，同时向源库中循环插入100000条记录，同步结束耗时78s，实例起了48个，从而推测上面的问题答案应该是启动当前时间周期的实例，而之前的忽略。但察看实例的启动时间，发现问题似乎并不完全这样，当时间间隔很小的事情下，quartz会连续把之前没启动的实例在某个周期内连续“补”回来，这一点让人费解，难道线程控制当真如此诡异？

4 楼 Godlikeme 2008-02-13  

可以控制job非并发执行，参考 javadoc org.quartz.StatefulJob。

3 楼 jones 2008-02-13  

嘿嘿，job实现StatefulJob接口就可以了，还是多看看API吧

2 楼 ddandyy 2008-02-13  

引用

quartz是一个按照设定的时间规则来调度作业的调度器，比如可以设定每30s启动一个Job，但如果Job在30s内还未完成，那么quartz默认情况下还是按照设定的周期启动新的Job线程。

这个应该是一个病句
就我说知quartz的两种调用方式 一种定时运行 一种循环运行
定时运行不考虑前一个是否执行完毕  循环运行当前一个没执行完的情况下  是不会启动下一个job的

两年前专门test过这个东西......现在应该不会变吧.....这种基础的东西

1 楼 抛出异常的爱 2008-02-13  

事务的颗度太大是你的问题根源....
你把所有的1000条数据都读出来了...
如果你不是一次读出所有的数据
那么你的问题是很简单的.
当第二个quartz起来时把前一个业务结束.