java实现高性能的数据同步

     最近在做一个银行的生产数据脱敏系统，今天写代码时遇到了一个“瓶颈”，脱敏系统需要将生产环境上Infoxmix里的数据原封不动的Copy到另一台 Oracle数据库服务器上，然后对Copy后的数据作些漂白处理。为了将人为干预的因素降到最低，在系统设计时采用Java代码对数据作Copy,思路 如图：



    首 先在代码与生产库间建立一个Connection，将读取到的数据放在ResultSet对象，然后再与开发库建立一个Connection。从 ResultSet取出数据后通过TestConnection插入到开发库，以此来实现Copy。代码写完后运行程序，速度太慢了，一秒钟只能Copy 一千条数据，生产库上有上亿条数据，按照这个速度同步完要到猴年马月呀,用PreparedStatement批处理速度也没有提交多少。我想能不能用多 线程处理，多个人干活总比一个人干活速度要快。
    假设生产库有1万条数据，我开5个线程，每个线程分2000条数据，同时向开发库里插数据，Oracle支持高并发这样的话速度至少会提高好多倍，按照这 个思路重新进行了编码，批处理设置为1万条一提交，统计插入数量的变量使用 java.util.concurrent.atomic.AtomicLong，程序一运行，传输速度飞快CPU利用率在70%~90%，现在一秒钟可 以拷贝50万条记录，没过几分钟上亿条数据一条不落地全部Copy到目标库。

在查询的时候我用了如下语句

String queryStr = "SELECT * FROM xx";
ResultSet coreRs = PreparedStatement.executeQuery(queryStr);

实习生问如果xx表里有上千万条记录，你全部查询出来放到ResultSet, 那内存不溢出了么？Java在设计的时候已经考虑到这个问题了，并没有查询出所有的数据，而是只查询了一部分数据放到ResultSet,数据“用完”它 会自动查询下一批数据，你可以用setFetchSize(int rows)方法设置一个建议值给ResultSet,告诉它每次从数据库Fetch多少条数据。但我不赞成，因为JDBC驱动会根据实际情况自动调整 Fetch的数量。另外性能也与网线的带宽有直接的关系。
相关代码

package com.dlbank.domain;  
  
import java.sql.Connection;  
import java.sql.PreparedStatement;  
import java.sql.ResultSet;  
import java.sql.Statement;  
import java.util.List;  
import java.util.concurrent.atomic.AtomicLong;  
  
import org.apache.log4j.Logger;  
  
/** 
 *<p>title: 数据同步类 </p>   
 *<p>Description: 该类用于将生产核心库数据同步到开发库</p>   
 *@author Tank Zhang  
 */  
public class CoreDataSyncImpl implements CoreDataSync {  
      
    private List<String> coreTBNames; //要同步的核心库表名  
    private ConnectionFactory connectionFactory;  
    private Logger log = Logger.getLogger(getClass());  
      
    private AtomicLong currentSynCount = new AtomicLong(0L); //当前已同步的条数  
      
    private int syncThreadNum;  //同步的线程数  
  
    @Override  
    public void syncData(int businessType) throws Exception {  
          
        for (String tmpTBName : coreTBNames) {  
            log.info("开始同步核心库" + tmpTBName + "表数据");  
            // 获得核心库连接  
            Connection coreConnection = connectionFactory.getDMSConnection(4);  
            Statement coreStmt = coreConnection.createStatement();  
            //为每个线程分配结果集  
            ResultSet coreRs = coreStmt.executeQuery("SELECT count(*) FROM "+tmpTBName);  
            coreRs.next();  
            //总共处理的数量  
            long totalNum = coreRs.getLong(1);  
            //每个线程处理的数量  
            long ownerRecordNum =(long) Math.ceil((totalNum / syncThreadNum));   
            log.info("共需要同步的数据量："+totalNum);  
            log.info("同步线程数量："+syncThreadNum);  
            log.info("每个线程可处理的数量："+ownerRecordNum);  
            // 开启五个线程向目标库同步数据  
            for(int i=0; i < syncThreadNum; i ++){  
                StringBuilder sqlBuilder = new StringBuilder();  
                //拼装后SQL示例  
                //Select * From dms_core_ds Where id between 1 And 657398  
                //Select * From dms_core_ds Where id between 657399 And 1314796  
                //Select * From dms_core_ds Where id between 1314797 And 1972194  
                //Select * From dms_core_ds Where id between 1972195 And 2629592  
                //Select * From dms_core_ds Where id between 2629593 And 3286990  
                //..  
                sqlBuilder.append("Select * From ").append(tmpTBName)  
                        .append(" Where id between " ).append(i * ownerRecordNum +1)  
                        .append( " And ")  
                        .append((i * ownerRecordNum + ownerRecordNum));  
                Thread workThread = new Thread(  
                        new WorkerHandler(sqlBuilder.toString(),businessType,tmpTBName));  
                workThread.setName("SyncThread-"+i);  
                workThread.start();  
            }  
            while (currentSynCount.get() < totalNum);  
            //休眠一会儿让数据库有机会commit剩余的批处理(只针对JUnit单元测试,因为单元测试完成后会关闭虚拟器，使线程里的代码没有机会作提交操作);  
            //Thread.sleep(1000 * 3);  
            log.info( "核心库"+tmpTBName+"表数据同步完成，共同步了" + currentSynCount.get() + "条数据");  
        }  
    }// end for loop  
      
    public void setCoreTBNames(List<String> coreTBNames) {  
        this.coreTBNames = coreTBNames;  
    }  
  
    public void setConnectionFactory(ConnectionFactory connectionFactory) {  
        this.connectionFactory = connectionFactory;  
    }  
      
    public void setSyncThreadNum(int syncThreadNum) {  
        this.syncThreadNum = syncThreadNum;  
    }  
      
    //数据同步线程  
    final class WorkerHandler implements Runnable {  
        ResultSet coreRs;  
        String queryStr;  
        int businessType;  
        String targetTBName;  
        public WorkerHandler(String queryStr,int businessType,String targetTBName) {  
            this.queryStr = queryStr;  
            this.businessType = businessType;  
            this.targetTBName = targetTBName;  
        }  
        @Override  
        public void run() {  
            try {  
                //开始同步  
                launchSyncData();  
            } catch(Exception e){  
                log.error(e);  
                e.printStackTrace();  
            }  
        }  
        //同步数据方法  
        void launchSyncData() throws Exception{  
            // 获得核心库连接  
            Connection coreConnection = connectionFactory.getDMSConnection(4);  
            Statement coreStmt = coreConnection.createStatement();  
            // 获得目标库连接  
            Connection targetConn = connectionFactory.getDMSConnection(businessType);  
            targetConn.setAutoCommit(false);// 设置手动提交  
            PreparedStatement targetPstmt = targetConn.prepareStatement("INSERT INTO " + targetTBName+" VALUES (?,?,?,?,?)");  
            ResultSet coreRs = coreStmt.executeQuery(queryStr);  
            log.info(Thread.currentThread().getName()+"'s Query SQL::"+queryStr);  
            int batchCounter = 0; //累加的批处理数量  
            while (coreRs.next()) {  
                targetPstmt.setString(1, coreRs.getString(2));  
                targetPstmt.setString(2, coreRs.getString(3));  
                targetPstmt.setString(3, coreRs.getString(4));  
                targetPstmt.setString(4, coreRs.getString(5));  
                targetPstmt.setString(5, coreRs.getString(6));  
                targetPstmt.addBatch();  
                batchCounter++;  
                currentSynCount.incrementAndGet();//递增  
                if (batchCounter % 10000 == 0) { //1万条数据一提交  
                    targetPstmt.executeBatch();  
                    targetPstmt.clearBatch();  
                    targetConn.commit();  
                }  
            }  
            //提交剩余的批处理  
            targetPstmt.executeBatch();  
            targetPstmt.clearBatch();  
            targetConn.commit();  
            //释放连接   
            connectionFactory.release(targetConn, targetPstmt,coreRs);  
        }  
    }  
}  

评论

35 楼 gaobo424 2010-11-26  

你是在哪家银行哦，有点搞扯，居然允许你连接它的核心数据库？？？你要是发一条Update语句呢？我做银行的项目都N久了，做过建行、商业银行的项目，谁敢把核心数据给你直连？？？顶多只能给你备过份出来，你再导入还差不多

34 楼 zhys513 2010-11-26  

迁移数据的话，PB也有个可以做不同数据库的对接（数据管道）。迁移不是很快，但操作很简单就是了。

33 楼 zhys513 2010-11-26  

jiayj198609 写道

yjwxfpl 写道

PreparedStatement targetPstmt = targetConn.prepareStatement("INSERT INTO " + targetTBName+" VALUES (?,?,?,?,?)");

你同步的表结构都只有五个字段  ,targetPstmt.setString()  类型全都是String？

那具体的表结构当然是看你具体的操作了；我的表只有五个字段；当然就set五个了；难道还要set6个？

用户set和用SQL拼接是有性能的差别的！既然是考虑大数据的迁移，所以对每条语句的性能都是要考虑到滴。

32 楼 LikeEJB_CC 2010-11-26  

一次1000条，还多线程，IO怎么解决
频繁的少量写入,不如一次大批量的写入.

31 楼 sunlong 2010-11-26  

线程处理上并不是很好吧，比如while (currentSynCount.get() < totalNum);你这是让主线程运行空语句，太浪费了

30 楼 smithfox 2010-11-26  

首先, 如果楼主真的是转载别人的文章, 需要加引说原文. 这种直接抄袭的做法, 我们的robbin哥是最鄙视的

yizhilong28 写道

1) CLOB/BLOB字段
2) 一次性查询过多数据, 遇到 "ORA-1013", "ORA-1555"，或堆栈溢出
3) insert时产生主键冲突

做过一同步的项目，1,2,3都遇到了，用java基本没解，头疼

这种数据库的手动复制用在一次性工作善可, 如果是一个长期使用的tools或是服务, 是很成问题的.

我所做的系统遇到了很多的问题, 主要有

1) CLOB/BLOB字段, 只比较长度, 不比较实际内容, 并且用ojdbc6.jar(最新oracle驱动fix了不少bug, 并且支持JDBC4标准, 所以用CLOB/ BLOG字段的支持更好

2) 一次性查询过多数据, 遇到 "ORA-1013", "ORA-1555": 需要对查询进行分段(分段逻辑需要针对不同的数据库定制, 一般情况是对类似lastmodifiedtime字段, 以时间分段

3) insert产生冲突, 或是其他的异常, 需要有非常好的异常处理框架, 因为JDBC的异常只有SQLException, 不能区分是连接异常, 表级异常, 还是行级异常. 并且要做好出异常的尝试机制. 只有这样才能避免因为中间异常而出现的中途退出和进入死循环.  要知道真正的产品上一次复制可能是好几天, 没有人能一直盯着看. 程序必须有很好的健壮性.

4) 对第三的补充, 尽管JDBC4标准已经对SQLException做了很好的改进, 但是各大数据库产商支持太差.

29 楼 吾日三省吾身 2010-11-26  

尊重人家版权嘛 转载的时候要说明哦

28 楼 sheep3600 2010-11-26  

codermouse 写道

最近刚在做一个同步项目，和楼主要求的很像。思路也和楼主要求的差不多。
不过说实话，不赞成用JAVA做数据同步，我再做的时候就出现很多不方便处理的问题，总是通过绕来绕去来解决。
可惜本人只会JAVA（而且还不是精通），其它的都不会，DBLINK也是开始做这个同步项目后才知道的，不过DBLINK用了一段时间，觉得很不稳定。也不赞成用。
如果有DBA，或者我是DBA，我想数据同步是个很容易解决的问题。
哎。。。感叹程序员要学的东西太多。
PS1:这是转贴？
PS2:JAVAEYE重新开张，数据丢失了，怎么还要我做发帖测试？还是经过前2天的事故，要求更进一步了？
PS3:恭喜JAVAEYE重新开张，不好意思，话有点过多了。

这种事情DBA估计不到10分钟就全搞定了，其实这种事情就该DBA去搞，各有所属，各有专攻嘛。
生产库别说select了，连ip我们都不知道，所以这种事还是不掺和了。

27 楼 renpeng301 2010-11-26  

sunnylocus 写道

jiayj198609 写道

imacback 写道

为什么要用java导呢，有没有别的解决方案？？？这个实在是慢，搞数据仓库的不错。

呵呵。。。解决方案有很多的；我也是实际项目赶到这儿了；所以就用java线程模式处理了！

老兄，你怎么随便拿我的文章到处贴，您也在大连银行工作？不过我好象没有见过你

真的是转的啊

26 楼 85977328 2010-11-26  

直接用dblink

25 楼 bitray 2010-11-26  

sunnylocus 写道

jiayj198609 写道

imacback 写道

为什么要用java导呢，有没有别的解决方案？？？这个实在是慢，搞数据仓库的不错。

呵呵。。。解决方案有很多的；我也是实际项目赶到这儿了；所以就用java线程模式处理了！

老兄，你怎么随便拿我的文章到处贴，您也在大连银行工作？不过我好象没有见过你

哈，直接遇到原作者就太尴尬了，不过老兄你也别介意。现在到处都是那种超级年轻的软件工程师。总的给他们一点点信心啊，哈哈

24 楼 xhdwell 2010-11-26  

用jdbc效率肯定不行的。建议通过动态组建脚本，使用脚本执行COPY

23 楼 pikenlike_123 2010-11-25  

我记得有个数据抽取/数据转换工具叫 ODI  
      ODI最大的特点是特征是提出了知识模块的概念(Knowledge Module)。
   可以百度一下..

22 楼 houzidexinsheng 2010-11-25  

去年我写了一个应用系统，不过场景和你有区别，我们主要是要同步基础数据，对需要同步的基础数据做版本控制，发布的时候直接把开发库的基础数据增量更新到生产库，如果需要的话，可以交流一下！

21 楼 yizhilong28 2010-11-25  

1) CLOB/BLOB字段
2) 一次性查询过多数据, 遇到 "ORA-1013", "ORA-1555"，或堆栈溢出
3) insert时产生主键冲突

做过一同步的项目，1,2,3都遇到了，用java基本没解，头疼

20 楼 gwpking8419 2010-11-25  

ETL工具呗

19 楼 codermouse 2010-11-25  

最近刚在做一个同步项目，和楼主要求的很像。思路也和楼主要求的差不多。
不过说实话，不赞成用JAVA做数据同步，我再做的时候就出现很多不方便处理的问题，总是通过绕来绕去来解决。
可惜本人只会JAVA（而且还不是精通），其它的都不会，DBLINK也是开始做这个同步项目后才知道的，不过DBLINK用了一段时间，觉得很不稳定。也不赞成用。
如果有DBA，或者我是DBA，我想数据同步是个很容易解决的问题。
哎。。。感叹程序员要学的东西太多。
PS1:这是转贴？
PS2:JAVAEYE重新开张，数据丢失了，怎么还要我做发帖测试？还是经过前2天的事故，要求更进一步了？
PS3:恭喜JAVAEYE重新开张，不好意思，话有点过多了。

18 楼 jiasky 2010-11-25  

你的这种思路人们用来做了很多事，
1个西瓜1个人吃了1分钟，
现在分成10个人，每个人6秒钟就吃完了，呵呵...

17 楼 icanfly 2010-11-25  

如果，我是说如果，你在中转复制的时候突然你的中转器（java实现的）挂了，当你恢复中转的时候，怎么处理？有容错吗？

16 楼 liliugen 2010-11-25  

天~\(≧▽≦)/~啦啦啦，这是银行系统啊 。关链接都不在finally里面的啊，这样的银行系统敢用么。。