問題描述

業(yè)務(wù)：需要從一個(gè)數(shù)據(jù)庫查詢百萬級數(shù)據(jù)，在java程序中插入到另一個(gè)oracle數(shù)據(jù)庫中代碼：

private final int persize = 1000; /** * 推送數(shù)據(jù)-流程 * @param tableCode 表名 * @param startTime 開始時(shí)間 * @param endTime 結(jié)束時(shí)間 * @return */ public boolean pushFrData(String username,String tableCode,String tableName,String startTime,String endTime){SimpleDateFormat sdf = new SimpleDateFormat('yyyy-MM-dd');System.out.println('導(dǎo)入數(shù)據(jù)到名錄庫！');boolean boo = false;//表名集合String [] str = tableCode.split(',');String [] names = tableName.split(',');startTime = startTime==''?'2000-01-01':startTime;endTime = endTime==''?getCurrentDate():endTime;//System.out.println('service 時(shí)間 '+startTime+' > '+endTime);String start_Time = 'to_date(’'+startTime+'’,’%Y-%m-%d’)';String end_Time = 'to_date(’'+endTime+'’,’%Y-%m-%d’)';System.out.println('選擇推送 '+str.length+' 張表');//遍歷表名集合for(int i = 0;i<str.length;i++){ System.out.println('所選數(shù)據(jù)表： '+str[i]); //字段集合 String [] arr = fillService.getIdenCode(str[i]); String iden_code = ''; //遍歷字段 for(int j = 0;j<arr.length;j++){iden_code += ''+arr[j]+','; } //表字段 iden_code = iden_code.substring(0,iden_code.length()-1); //System.out.println(str[i]+'總共 '+arr.length+' 個(gè)字段!'); //得到推送數(shù)據(jù)集合 String sql = 'select count(*) from '+str[i] +' where s_ext_timestamp >= '+start_Time+' and s_ext_timestamp < '+end_Time; System.out.println(sql); int table_size = Integer.valueOf(frDao.query(sql).get(0).toString()); //System.out.println(table_size/persize); int times = table_size%persize==0?table_size/persize:table_size/persize+1; for(int t = 1; t <= times;t++){int start = (t-1) * persize;List<Object[]> lists = getData(str[i], iden_code,startTime,endTime,start);//推送數(shù)據(jù)System.out.println('準(zhǔn)備導(dǎo)出第 '+t+' 批數(shù)據(jù)');push(lists,str[i],iden_code);System.out.println('已導(dǎo)出第 '+t+' 批數(shù)據(jù)'); } try{if(table_size > 0){ addLog(username,str[i].toString(),names[i].toString(),table_size,'1',sdf.parse(startTime),sdf.parse(endTime));} }catch (ParseException e){System.out.println('日期格式轉(zhuǎn)換異常'); }}return boo; } /** * 推送數(shù)據(jù) * @param lists 數(shù)據(jù)集 * @param table 表名 * @param iden_code 字段集 */ private int push(List<Object []> lists,String table,String iden_code){boolean boo = false;int count = 0;//遍歷數(shù)據(jù)結(jié)果集if(lists.size()>0){ for(int k = 0;k < lists.size();k++){Object [] obj = lists.get(k);String val = ''; //將數(shù)據(jù)轉(zhuǎn)換成String類型for(Object s:obj){ //val += '’'+s.toString()+'’,'; if(s != null){val += '’'+s.toString()+'’|'; }else{val += '’’|'; }}val = val.substring(0,val.length()-1);String etpsid = '';String [] iden = iden_code.split(',');String [] value = val.split('|');String val2 = '';//格式化數(shù)據(jù)（日期）if(iden.length == value.length){ //格式化sql語句的時(shí)間 for(int i = 0;i<iden.length;i++){//判斷字段是否是date類型if(getDateType(table, iden[i])){ //格式化字符串 防止出現(xiàn)datetime類型 1900-01-01 00:00:00.0的情況 if(value[i].length() > 4){String vv = value[i].substring(value[i].length()-3,value[i].length());if(vv.contains('.')){ val2 += 'to_date('+value[i].substring(0,value[i].length()-3)+'’,’YYYY-MM-DD HH24:MI:SS’),';}else{ val2 += 'to_date('+value[i]+',’YYYY-MM-DD HH24:MI:SS’),';} }else{val2 += '’’,'; }}else{ val2 += value[i]+',';}if('ETPSID'.equals(iden[i])){ etpsid = value[i];} } val2 = val2.substring(0,val2.length()-1);}else{ /*System.out.println(iden.length+' : '+value.length); for(int j = 0 ;j< value.length;j++){System.out.println(value[j]); }*/ System.out.println('推送數(shù)據(jù)和字段不一致');}String mlk_table = getMlkTable(table);String mlk_code = getMlkCode(iden_code,table);//插入數(shù)據(jù)sql//String sql = 'insert into '+table+'('+iden_code+') values ('+val2+')';//生成流水號String uuid = UUID.randomUUID().toString();uuid = uuid.replace('-','');val2 += ',’'+uuid+'’';String sql = 'insert into '+mlk_table+'('+mlk_code+') values ('+val2+')';if(etpsid != ''){ //System.out.println(etpsid); //工商增量數(shù)據(jù)按日依'企業(yè)唯一標(biāo)識'為關(guān)鍵字，更新、追加至名錄庫表中 String s_sql = 'select * from '+mlk_table+' where 企業(yè)唯一標(biāo)識 = '+etpsid; //System.out.println(s_sql); int s = mlDao.query(s_sql).size(); if (s > 0){String update_sql = '';String [] update_code = mlk_code.split(',');for(int j = 0;j < iden_code.split(',').length;j++){ //判斷字段是否是date類型 if(getDateType(table, iden[j])){//格式化字符串 防止出現(xiàn)datetime類型 1900-01-01 00:00:00.0的情況if(value[j].length() > 4){ String vv = value[j].substring(value[j].length() - 3, value[j].length()); if(vv.contains('.')){update_sql += update_code[j]+' = '+ 'to_date('+value[j].substring(0, value[j].length() - 3)+'’,’YYYY-MM-DD HH24:MI:SS’),'; }else{update_sql += update_code[j]+' = '+ 'to_date('+value[j]+',’YYYY-MM-DD HH24:MI:SS’),'; }}else{ update_sql += update_code[j]+' = '+ '’’,';} }else{update_sql += update_code[j]+' = '+ value[j]+','; } //update_sql += update_code[j]+' = '+update_val[j]+',';}update_sql += '流水號 = ’'+uuid+'’';update_sql = 'update '+mlk_table+' set '+update_sql+' where 企業(yè)唯一標(biāo)識 = '+etpsid;//System.out.println('/////////// /n'+update_sql+'n');sql = update_sql;System.out.println('更新 '+mlk_table+' 標(biāo)識 '+etpsid); }else{System.out.println('插入 '+mlk_table+' 標(biāo)識 '+etpsid); }}//System.out.println(sql);try{ //循環(huán)執(zhí)行sql mlDao.execute(sql); count++; System.out.println(table+' 推送第 '+count+' 條');}catch (Exception e){ System.out.println('sql執(zhí)行異常！');} } System.out.println(table+' 表共插入 '+count+' 條數(shù)據(jù)!');}else{ System.out.println(table+' 表共插入 '+count+' 條數(shù)據(jù)!'); System.out.println('導(dǎo)入數(shù)據(jù)為空！');}return count; }

問題解答

讀取的話可以多線程讀取，插入的話看以下鏈接：

https://segmentfault.com/sear...

https://segmentfault.com/sear...

1、數(shù)據(jù)保存到數(shù)據(jù)庫可以使用批處理比如一次處理100條插入記錄2、不要一次性處理所有數(shù)據(jù)，可以將1W條數(shù)據(jù)分成兩個(gè)線程來處理，這樣可以充分利用cpu，同事不會(huì)導(dǎo)致大的阻塞

建議使用JDBC的批處理模式，搜索以下關(guān)鍵字：addBatch(String query)executeBatch()

建議1000條左右作為一個(gè)batch提交事務(wù)。

懶人方法:

在目標(biāo)oracle數(shù)據(jù)庫中建立分布式鏈接,直接insert into 目標(biāo)表 select sql語句