著录数据打包

使用 MyBatis 连接 Oracle ，然后向 MongoDB 中写入。

数据量 1.3 亿，每分钟 7 - 10 万。

SyncPacker_201603.java

package syncPacker;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.http.message.BasicNameValuePair;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.Mongo;
import com.pro.framework.action.BaseController;

import logCenter.SendLog;
import net.sf.json.JSONObject;
import syncPacker.bean.PatentBibliographicChangeBean;
import syncPacker.bean.SyncDataPackageBean;
import utils.DateUtils;
import utils.DatetimeUtils;
import utils.HttpUtils;

/**
 * 增量数据分块打包 全处理
 * 
 * http://localhost:8080/PatentSearchExtend/syncPacker!pack.action
 * 
 * http://10.78.2.21:8080/PatentSearchExtend/syncPacker!pack.action?bean.
 * tableName=E_BIBLIOGRAPHIC_CHANGE_TEMP&bean.maxRowsPerSyncPackerPackage= 10000
 */
@Service
public class SyncPacker_201603 extends BaseController {

    private static final long serialVersionUID = 1L;

    // 初始化：数据库接口
    @Autowired
    private SyncPackerDao dao;

    // 初始化：发送端地址
    @Value("${url_syncSender}")
    private String url_syncSender;

    // 初始化：本地文件存储路径
    @Value("${path_syncPacker_package}")
    private String path_syncPacker_package;

    // 初始化：读取最大数据包名称的地址
    @Value("${url_selectMaxPackageNumber}")
    private String url_selectMaxPackageNumber;

    // 初始化：存储最大数据包名称的地址
    @Value("${url_insertSyncDataPackage}")
    private String url_insertSyncDataPackage;

    // 初始化：查询条件Bean
    private SyncDataPackageBean bean = new SyncDataPackageBean();
    // 初始化：查询结果List
    private List<PatentBibliographicChangeBean> pbcList = new ArrayList<PatentBibliographicChangeBean>();
    // 初始化：形成的数据包名称
    private List<PatentBibliographicChangeBean> pbcList_withPackageName = new ArrayList<PatentBibliographicChangeBean>();
    // 初始化：已打包的增量数据ID表单
    private List<String> pdaIdList = new ArrayList<String>();
    // 初始化：用于删除数据的临时ID List
    private List<String> idPartList = new ArrayList<String>();
    // 初始化：传输协议
    private HttpUtils httpUtils = new HttpUtils();
    // 初始化：键值串
    private List<BasicNameValuePair> paramList = new ArrayList<BasicNameValuePair>();
    // POST 结果
    private String str_postResult;
    // 初始化：发送url返回的信息
    private JSONObject json_postResult;
    // 初始化：历史最大包编号
    private String maxPackageNumber;
    // 初始化：记录打包完成后的数据版本
    private Integer centerNodeDataVersion;
    // 发送远程日志
    private SendLog sendLog = new SendLog();
    // 记录本地日志
    private Logger logger = Logger.getLogger(SyncPacker_201603.class);
    // 初始化：判断程序是否正在运行
    public static boolean isRunning = false;

    // 本次处理完成后的最大包编号
    private String packedPackageNumber;
    // 用于返回json的成功信息
    private String success = "success";
    

    /** 文件补发用：指定数据重新打包 */
    // http://localhost:8080/PatentSearchExtend/syncPacker!packByPackageNumber.action?bean.packageNumberStart=000101&bean.packageNumberEnd=000102
    public String packByPackageNumber() throws Exception {

        logMemory("本次请求处理开始。", bean.getPackageNumberStart() + " - " + bean.getPackageNumberEnd() );

        String job_start = DateUtils.getCurrentTimeString(0);
        
        try {
            

            for ( int i = Integer.valueOf(bean.getPackageNumberStart()); 
                    i <= Integer.valueOf(bean.getPackageNumberEnd()); i++) {                

                // 开始时间
                String package_start = DateUtils.getCurrentTimeString(0);
                
                // 包编号
                String packageNumber = String.format( "%06d", i );
                
                logMemory("开始，包编号：", packageNumber );
                
                //（1）读历史表
                pbcList = selectList_changeHistory( packageNumber ); 
                
                if( null == pbcList ) {
                    logMemory("<span style="color:red;">数据为空 ！！！</span>", "");
                    
                } else{
                    logMemory("数据查询完毕，数据量为", String.valueOf( pbcList.size() ));

                    //（2）插入MongoDB
                    insertMongoDB( pbcList );
                    pbcList.clear();
                };
                
                logMemory("传输结束，包编号：" + packageNumber ," 用时: "
                        + DatetimeUtils.getDistanceTimes_string(package_start, DateUtils.getCurrentTimeString(0)) );
            }
        } catch (Exception e) {
            // 日志输出
            logMemory("系统发生异常", e.getMessage());
            e.printStackTrace();
        }
        
        logMemory("本次请求处理完成，程序结束。", bean.getPackageNumberStart() + " - " + bean.getPackageNumberEnd()
                + " 用时: " + DatetimeUtils.getDistanceTimes_string( job_start, DateUtils.getCurrentTimeString(0)) );
        
        return SUCCESS;
    }
    

    /**
     * 读历史表
     * @param packageNumber
     * @return
     */
    private List<PatentBibliographicChangeBean> selectList_changeHistory( String packageNumber ){
        
        for ( int i = 0; i < 100; i++ ) {
            
            try {
                return dao.selectList_changeHistory( packageNumber );                
            } catch (Exception e) {
                // TODO Auto-generated catch block
//                e.printStackTrace();
                logMemory("系统发生异常", e.getMessage());
                try {
                    Thread.sleep(500);
                    logMemory("暂停 0.5 秒", "" );
                } catch (InterruptedException e1) {
                    // TODO Auto-generated catch block
                    e1.printStackTrace();
                }
            } 
        }// loop end
        return null ;
        
    }
    
    
    

    /**
     * 插入 MongoDB
     */
    public void insertMongoDB( List<PatentBibliographicChangeBean> pbcList ) {    

        //# MongoDB(数据加载目标)
        String syncLoadIntoMongoDbService = "10.78.2.23:27017";
        String syncLoadIntoMongoDbName = "patent_search_extend";
        String syncLoadIntoMongoTable = "patent_bibliographic_20160319";

        // 加载开始
//        logger.info(DateUtils.getNow() + " Load start: " + syncLoadIntoMongoDbService );

        Mongo m = null;
        try {
            m = new Mongo( syncLoadIntoMongoDbService );
        } catch (UnknownHostException e) {
            e.printStackTrace();
            logger.info(DateUtils.getNow() + " UnknownHostException:" + e.getMessage());
        }

        // 库名
        DB db = m.getDB( syncLoadIntoMongoDbName );
//        logger.info( DateUtils.getNow() + " Db:" + syncLoadIntoMongoDbName );

        // 表名
        DBCollection collection = db.getCollection( syncLoadIntoMongoTable );
//        logger.info(DateUtils.getNow() + " Table:" + syncLoadIntoMongoTable );
        
        // 循环列表，将每个元素插入数据库
        for( PatentBibliographicChangeBean pbcBean : pbcList ){        
            
            
            
            //（1）读取一条著录数据
//            JSONObject json = packByPackageNumber_readBibl( pbcBean.getId() );
            
//            if( null == json ) return ;

            // 列，值
            BasicDBObject insDoc = new BasicDBObject();
            
            insDoc.put("abstract_No"    , pbcBean.getAbstract_No()    );
            insDoc.put("app_Addr"       , pbcBean.getApp_Addr()       );
            insDoc.put("app_Cn"         , pbcBean.getApp_Cn()         );
            insDoc.put("app_Country"    , pbcBean.getApp_Country()    );
            insDoc.put("app_Date"       , pbcBean.getApp_Date()       );
            insDoc.put("app_Name"       , pbcBean.getApp_Name()       );
            insDoc.put("app_Sn"         , pbcBean.getApp_Sn()         );
            insDoc.put("app_Type"       , pbcBean.getApp_Type()       );
            insDoc.put("app_Zip"        , pbcBean.getApp_Zip()        );
            insDoc.put("ecla"           , pbcBean.getEcla()           );
            insDoc.put("fi"             , pbcBean.getFi()             );
            insDoc.put("ft"             , pbcBean.getFt()             );
            insDoc.put("id"             , pbcBean.getId()             );
            insDoc.put("inv_Title"      , pbcBean.getInv_Title()      );
            insDoc.put("invent_Type"    , pbcBean.getInvent_Type()    );
            insDoc.put("inventor"       , pbcBean.getInventor()       );
            insDoc.put("ipc_Standard"   , pbcBean.getIpc_Standard()   );
            insDoc.put("locarno"        , pbcBean.getLocarno()        );
            insDoc.put("operation_Time" , pbcBean.getOperation_Time() );
            insDoc.put("operation_Type" , pbcBean.getOperation_Type() );
            insDoc.put("package_Name"   , pbcBean.getPackage_Name()   );
            insDoc.put("pct_App_Cn"     , pbcBean.getPct_App_Cn()     );
            insDoc.put("pct_App_Date"   , pbcBean.getPct_App_Date()   );
            insDoc.put("pct_App_Sn"     , pbcBean.getPct_App_Sn()     );
            insDoc.put("pct_Date"       , pbcBean.getPct_Date()       );
            insDoc.put("pct_Pub_Cn"     , pbcBean.getPct_Pub_Cn()     );
            insDoc.put("pct_Pub_Date"   , pbcBean.getPct_Pub_Date()   );
            insDoc.put("pct_Pub_Lang"   , pbcBean.getPct_Pub_Lang()   );
            insDoc.put("pct_Pub_Sn"     , pbcBean.getPct_Pub_Sn()     );
            insDoc.put("prn"            , pbcBean.getPrn()            );
            insDoc.put("prn_Cn"         , pbcBean.getPrn_Cn()         );
            insDoc.put("prn_Date"       , pbcBean.getPrn_Date()       );
            insDoc.put("prn_Sn"         , pbcBean.getPrn_Sn()         );
            insDoc.put("prn_Type"       , pbcBean.getPrn_Type()       );
            insDoc.put("pub_Cn"         , pbcBean.getPub_Cn()         );
            insDoc.put("pub_Date"       , pbcBean.getPub_Date()       );
            insDoc.put("pub_Sn"         , pbcBean.getPub_Sn()         );
            insDoc.put("pub_Type"       , pbcBean.getPub_Type()       );
            insDoc.put("uc"             , pbcBean.getUc()             );
            
            collection.insert(insDoc);
            insDoc.clear();
            
        }

        // 循环遍历pdaBeanList
//        System.out.println("loading ...");
//        logger.info(DateUtils.getNow() + " rows:" + pdaBeanList.size());

        // 当前记录编号
        // int currentRowNumber = 0;

        if ( m != null) m.close();
        
//        System.out.println("Load finished.");
    }
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    

    /** 记录日志 */
    private void logMemory(String behavior, String content) {
        // 向服务器发送日志
//        sendLog.send("syncPacker", behavior, content);
        // 记录本地日志
        logger.info(DateUtils.getNow() + " " + behavior + " :" + content);
        // 控制台输出日志
//        System.out.println("syncPacker : " + DateUtils.getNow() + " " + behavior + " :" + content);
    }

    @Override
    public String insert() throws Exception {
        return null;
    }

    @Override
    public String update() throws Exception {
        return null;
    }

    @Override
    public String selectList() throws Exception {
        return null;
    }

    @Override
    public String delete() throws Exception {
        return null;
    }

    public static boolean isRunning() {
        return isRunning;
    }

    public static void setRunning(boolean isRunning) {
        SyncPacker_201603.isRunning = isRunning;
    }

    public Integer getCenterNodeDataVersion() {
        return centerNodeDataVersion;
    }

    public void setCenterNodeDataVersion(Integer centerNodeDataVersion) {
        this.centerNodeDataVersion = centerNodeDataVersion;
    }

    public String getSuccess() {
        return success;
    }

    public void setSuccess(String success) {
        this.success = success;
    }

    public String getMaxPackageNumber() {
        return maxPackageNumber;
    }

    public void setMaxPackageNumber(String maxPackageNumber) {
        this.maxPackageNumber = maxPackageNumber;
    }

    public String getPackedPackageNumber() {
        return packedPackageNumber;
    }

    public void setPackedPackageNumber(String packedPackageNumber) {
        this.packedPackageNumber = packedPackageNumber;
    }

    public SyncDataPackageBean getBean() {
        return bean;
    }

    public void setBean(SyncDataPackageBean bean) {
        this.bean = bean;
    }
}

一个备份，记一下吧：

package syncPacker;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.http.message.BasicNameValuePair;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.Mongo;
import com.pro.framework.action.BaseController;

import logCenter.SendLog;
import net.sf.json.JSONObject;
import syncPacker.bean.PatentBibliographicChangeBean;
import syncPacker.bean.SyncDataPackageBean;
import utils.DateUtils;
import utils.DatetimeUtils;
import utils.HttpUtils;

/**
 * 增量数据分块打包 全处理
 * 
 * http://localhost:8080/PatentSearchExtend/syncPacker!pack.action
 * 
 * http://10.78.2.21:8080/PatentSearchExtend/syncPacker!pack.action?bean.
 * tableName=E_BIBLIOGRAPHIC_CHANGE_TEMP&bean.maxRowsPerSyncPackerPackage= 10000
 */
@Service
public class SyncPacker_201603 extends BaseController {

    private static final long serialVersionUID = 1L;

    // 初始化：数据库接口
    @Autowired
    private SyncPackerDao dao;

    // 初始化：发送端地址
    @Value("${url_syncSender}")
    private String url_syncSender;

    // 初始化：本地文件存储路径
    @Value("${path_syncPacker_package}")
    private String path_syncPacker_package;

    // 初始化：读取最大数据包名称的地址
    @Value("${url_selectMaxPackageNumber}")
    private String url_selectMaxPackageNumber;

    // 初始化：存储最大数据包名称的地址
    @Value("${url_insertSyncDataPackage}")
    private String url_insertSyncDataPackage;

    // 初始化：查询条件Bean
    private SyncDataPackageBean bean = new SyncDataPackageBean();
    // 初始化：查询结果List
    private List<PatentBibliographicChangeBean> pbcList = new ArrayList<PatentBibliographicChangeBean>();
    // 初始化：形成的数据包名称
    private List<PatentBibliographicChangeBean> pbcList_withPackageName = new ArrayList<PatentBibliographicChangeBean>();
    // 初始化：已打包的增量数据ID表单
    private List<String> pdaIdList = new ArrayList<String>();
    // 初始化：用于删除数据的临时ID List
    private List<String> idPartList = new ArrayList<String>();
    // 初始化：传输协议
    private HttpUtils httpUtils = new HttpUtils();
    // 初始化：键值串
    private List<BasicNameValuePair> paramList = new ArrayList<BasicNameValuePair>();
    // POST 结果
    private String str_postResult;
    // 初始化：发送url返回的信息
    private JSONObject json_postResult;
    // 初始化：历史最大包编号
    private String maxPackageNumber;
    // 初始化：记录打包完成后的数据版本
    private Integer centerNodeDataVersion;
    // 发送远程日志
    private SendLog sendLog = new SendLog();
    // 记录本地日志
    private Logger logger = Logger.getLogger(SyncPacker_201603.class);
    // 初始化：判断程序是否正在运行
    public static boolean isRunning = false;

    // 本次处理完成后的最大包编号
    private String packedPackageNumber;
    // 用于返回json的成功信息
    private String success = "success";

    /** 主处理：数据同步_增量数据分块打包 */
    // @Scheduled(cron = "${scheduled_syncPacker}")
    public String pack() {

        logMemory("scheduled_syncPacker start", "");
        if (isRunning) {
            logger.info(DateUtils.getNow() + "正在运行，退出");
            return SUCCESS;
        }
        isRunning = true;
        // 数据包数量
        int packagesAmount = 0;
        Date startTime = new Date();

        for (int i = 1; i <= 1; i++) {
            logMemory("循环内部（单个数据包）开始：内部编号", "" + i);

            try {
                // 1.读取定量数据
                selectPbcList();
                if (pbcList.size() <= 0) {
                    logMemory("无数据", "退出");
                    break;
                }

                // 2.形成文件
                if (save2disk()) {
                    packagesAmount++;
                    // 3.转移数据
                    transferFromChange2History();
                }
            } catch (Exception e) {
                // 日志输出
                logMemory("系统发生异常", e.getMessage());
                e.printStackTrace();
                break;
            } finally {
                // 清空临时List
                paramList.clear();
                pbcList.clear();
                pbcList_withPackageName.clear();
                pdaIdList.clear();
                System.gc();
            }
            logMemory("循环内部（单个数据包）结束，内部编号", "" + i);
        }
        // loop end

        // 清空临时List
        paramList.clear();
        pbcList.clear();
        pbcList_withPackageName.clear();
        pdaIdList.clear();
        System.gc();

        isRunning = false;
        logMemory("本次打包结束，新增数据包数量：" + packagesAmount + "，历时："
                + DatetimeUtils.getDistanceTimes_string(startTime, new Date()), "");
        logMemory("finished", "scheduled_syncPacker end");

        return SUCCESS;
    }

    /** selectPbcList() */
    // 读取 专利著录变化 数据 数据库有连接失败的时候，尝试 10 次
    private void selectPbcList() {

        // 日志输出
        logMemory("增量数据分块打包开始执行", "");

        for (int i = 0; i < 10; i++) {
            try {
                // -------------- 查询分块著录信息增量数据 --------------
                logMemory("准备读取数据", "");
                // 每次取10000
                pbcList = dao.selectList(bean);
                logMemory("读取完毕", "数据量" + pbcList.size());
                return;
            } catch (Exception e) {
                // 日志输出
                logMemory("系统发生异常", e.getMessage());
                e.printStackTrace();
            }
        }
        // loop end
        return;
    }

    /**
     * save2disk()
     * 
     * 将打包文件存盘
     */
    private boolean save2disk() throws Exception {

        // -------------- 分块打包增量数据 --------------

        // 查询上一次处理最后生成的数据包名称
        logMemory("准备查询最后一个数据包名称", "");

        // 发送url，请求对应服务器返回最大数据包名称
        // syncDataPackageBean = dao.lastDataPackageInfo(0);
        str_postResult = httpUtils.post(url_selectMaxPackageNumber, null);
        logMemory("取最大包编号：" + url_selectMaxPackageNumber, " 返回 " + str_postResult);

        // 判断是否成功发送信息
        if (str_postResult == null || "".equals(str_postResult)) {
            // 发送失败，跳出循环，结束程序
            logMemory("失败！", " 返回为空");
            isRunning = false;
            return false;
        }

        // 解析Post返回的Json
        json_postResult = JSONObject.fromObject(str_postResult);

        // 判断是否成功获取到最大包编号
        if ("success".equalsIgnoreCase(json_postResult.getString("success"))) {
            // 发送成功，进行下一步操作
            maxPackageNumber = json_postResult.get("maxPackageNumber").toString();
        } else {
            // 发送失败，跳出循环，结束程序
            logMemory("失败！", "没有返回成功标志。");
            isRunning = false;
            return false;
        }

        logMemory("上一次最后一个数据包名称", maxPackageNumber);
        logMemory("准备将数据写入文件", " 路径：" + path_syncPacker_package);

        // 得到新数据包名,同时分块打包增量数据
        pbcList_withPackageName = DataPacking_2.CompressPdaData(pbcList, maxPackageNumber, path_syncPacker_package);

        // 日志输出
        logMemory("新数据包名称", pbcList_withPackageName.get(0).getPackage_Name() + ".txt");
        logMemory("转换完成，数量", String.valueOf(pbcList_withPackageName.size()));

        // 记录打包完成的数据版本信息（流水号）
        centerNodeDataVersion = Integer.parseInt(pbcList_withPackageName.get(0).getPackage_Name());

        // -------------- 记录已完成的数据包名称 --------------
        // 发送打包完成的数据包名称（流水号）
        // dao.insertPacName(syncDataPackageBean);
        paramList = new ArrayList<BasicNameValuePair>();
        paramList.add(0,
                new BasicNameValuePair("packedPackageNumber", pbcList_withPackageName.get(0).getPackage_Name()));

        str_postResult = httpUtils.post(url_insertSyncDataPackage, paramList);
        logMemory("新增数据包：" + url_insertSyncDataPackage, " 返回 " + str_postResult);

        // 判断是否成功发送信息
        if (str_postResult == null || "".equals(str_postResult)) {
            // 发送失败，跳出循环，结束程序
            logMemory("失败！", " 返回为空");
            isRunning = false;
            return false;
        } else {
            // 解析Post返回的Json
            json_postResult = JSONObject.fromObject(str_postResult);
            // 判断是否成功发送信息
            if (!"success".equalsIgnoreCase(json_postResult.getString("success"))) {
                // 发送失败，跳出循环，结束程序
                logMemory("失败！", "没有返回成功标志。");
                isRunning = false;
                return false;
            }
        }
        logMemory("新增数据包，完成", "");
        return true;
    }

    /**
     * transferFromChange2History
     * 
     * 将数据从 Change 表移至 History 表
     */
    private void transferFromChange2History() throws Exception {

        // -------------- 添加数据至归档表 --------------
        logMemory("准备将已打包的记录移入归档表，数据量：", String.valueOf(pbcList_withPackageName.size()));

        // 根据打包后的全量数据量，循环执行插入历史表
        for (int j = 0; j < pbcList_withPackageName.size(); j++) {
            // 执行SQL插入操作语句
            dao.appendIntoChangeHistory(pbcList_withPackageName.get(j));
            pdaIdList.add(pbcList_withPackageName.get(j).getId());
        }
        // 日志输出
        logMemory("插入归档表，结束", String.valueOf(pdaIdList.size()));

        // -------------- 删除操作表的数据 --------------
        logMemory("准备删除专利著录信息变化表中的数据，数据量", String.valueOf(pdaIdList.size()));
        // List元素数量
        double total = pdaIdList.size();
        // 总SQL删除语句执行次数
        int loop = (int) Math.ceil(total / 1000);
        // List下标控制
        int idnumber;
        for (int k = 0; k < loop; k++) {
            // 每次只能删除 1000 条，否则SQL长度超长，出错。
            for (int z = 0; z < 1000; z++) {
                idnumber = k * 1000 + z;
                if (idnumber >= total) {
                    break;
                }
                // 循环拼接删除条件List
                idPartList.add(pbcList_withPackageName.get(idnumber).getId());
            }
            // 执行批量删除操作
            dao.removeFromChangeTable(idPartList);
            idPartList.clear();
        }
        // 日志输出
        logMemory("全部删除完毕,删除数据量：", String.valueOf(pdaIdList.size()));
    }

    /** 获取历史最大Json编号 */
    // http://10.78.2.22:8080/PatentSearchExtend/syncPacker!selectMaxPackageNum.action
    public String selectMaxPackageNum() {
        try {
            maxPackageNumber = dao.getMaxPackageNumber();
        } catch (Exception e) {
            e.printStackTrace();
            this.setSuccess(false);
        }
        return SUCCESS;
    }

    /** 存储当前打包Json编号 */
    // http://10.78.2.22:8080/PatentSearchExtend/syncPacker!insertSyncDataPackage.action
    public String insertSyncDataPackage() {
        try {
            dao.insertSyncDataPackage(packedPackageNumber);
        } catch (Exception e) {
            e.printStackTrace();
            this.setSuccess(false);
        }
        return SUCCESS;
    }

    /** 文件补发用：指定数据重新打包 */
    // http://localhost:8080/PatentSearchExtend/syncPacker!packByPackageNumber.action?bean.packageNumberStart=000101&bean.packageNumberEnd=000102
    public String packByPackageNumber() throws Exception {

        logMemory("本次请求处理开始。", bean.getPackageNumberStart() + " - " + bean.getPackageNumberEnd() );
        
        try {
            StringBuffer fileFullName;
            // 函数计算用Integer型数据
//            int packageNumberStart = Integer.valueOf(bean.getPackageNumberStart());
//            int packageNumberEnd = Integer.valueOf(bean.getPackageNumberEnd());
            // 重新设置bean地数据包编号
//            bean.setPackageNumberStart(String.format("%06d", packageNumberStart));
//            bean.setPackageNumberEnd(String.format("%06d", packageNumberEnd));
            // 输出日志
//            logMemory("本次手动补包的开始编号为", bean.getPackageNumberStart());
//            logMemory("本次手动补包的结束编号为", bean.getPackageNumberEnd());
            
            // 打包指定数据包编号对应的数据
//            for (int i = 0; i < Integer.valueOf(bean.getPackageNumberEnd())
//                    - Integer.valueOf(bean.getPackageNumberStart()) + 1; i++) {
            for ( int i = Integer.valueOf(bean.getPackageNumberStart()); 
                    i <= Integer.valueOf(bean.getPackageNumberEnd()); i++) {                

                String date_start = DateUtils.getCurrentTimeString(0);
                
                String packageNumber = String.format( "%06d", i );
                
                logMemory("开始，包编号：", packageNumber );
                
                // 删除现有文件
//                logMemory("开始配置数据包存储路径", "");
//                fileFullName = new StringBuffer();
////                fileFullName.append(path_syncPacker_package);
//                fileFullName.append("d:/Patent_Bibliographic_Change_Packages/PBC_");
//                fileFullName.append( packageNumber );
//                fileFullName.append(".txt");
//                logMemory("数据包全路径为", fileFullName.toString());
//                 
//                File file;
//                file = new File(fileFullName.toString());
//                if(file.exists()&&file.isFile()) file.delete();
                
//                if (file.isFile() && file.exists()) {
//                    logMemory("原数据包存在", fileFullName.toString());
//                    file.delete();
//                    logMemory("原数据包已删除", fileFullName.toString());
//                } else {
//                    logMemory("原数据包不存在，准备创建新文件", fileFullName.toString());
//                }
                
                // 打包数据
//                logMemory("设置新的文件包编号", String.format("%06d", "" + i));
//                bean.setPackageNumber(String.format("%06d", "" + i));
//                pbcList = dao.repackList(bean);
                
                pbcList = packByPackageNumber_readHistory( packageNumber ); 
                
                if( null == pbcList ) continue ;
                
                logMemory("数据查询完毕，数据量为", String.valueOf( pbcList.size() ));
                
//                Map<String, JSONObject> bibliographicMap = new HashMap();
//                FileWriter writer = new FileWriter( fileFullName.toString(), true);  
//                
//                for( PatentBibliographicChangeBean pbcBean : pbcList ){
//                    
//                    // 读取 著录 数据
//                    bibliographicMap.put( pbcBean.getId(), packByPackageNumber_readBibl( pbcBean.getId() ));
//                    try {  
//                        //打开一个写文件器，构造函数中的第二个参数true表示以追加形式写文件  
//                        writer.write( packByPackageNumber_readBibl( pbcBean.getId() ).toString() + '
' );  
//                    } catch (IOException e) {  
//                        e.printStackTrace();  
//                    } 
//                }
//                writer.close();  
                insertMongoDB( pbcList );
                
////                File file = new File(fileName); 
//                BufferedReader reader = null;  
//                try {  
//                    System.out.println("以行为单位读取文件内容，一次读一整行：");  
//                    reader = new BufferedReader(new FileReader(file));  
//                    String tempString = null;  
//                    int line = 1;  
//                    // 一次读入一行，直到读入null为文件结束  
//                    while ((tempString = reader.readLine()) != null) {  
//                        // 显示行号  
//                        System.out.println("line " + line + ": " + tempString);  
//                        line++;  
//                    }  
//                    reader.close();  
//                } catch (IOException e) {  
//                    e.printStackTrace();  
//                } finally {  
//                    if (reader != null) {  
//                        try {  
//                            reader.close();  
//                        } catch (IOException e1) {  
//                        }  
//                    }  
//                }  
                
//                for( String mapKey : bibliographicMap.keySet()){
//                    JSONObject bibliographic = bibliographicMap.get( mapKey );
//                    System.out.println( bibliographic );
//                }
                
//                logMemory("数据打包完毕数据", fileFullName.toString());
                pbcList.clear();
                
                logMemory(DateUtils.getNow() + " 传输结束，包编号：" + packageNumber ," 用时: "
                        + DatetimeUtils.getDistanceTimes_string(date_start, DateUtils.getCurrentTimeString(0)));
            }
        } catch (Exception e) {
            // 日志输出
            logMemory("系统发生异常", e.getMessage());
            e.printStackTrace();
        }
        logMemory("本次请求处理完成，程序结束。", bean.getPackageNumberStart() + " - " + bean.getPackageNumberEnd() );
        
        return SUCCESS;
    }
    

    /**
     * 读历史表
     * @param packageNumber
     * @return
     */
    private List<PatentBibliographicChangeBean> packByPackageNumber_readHistory( String packageNumber ){
        
        for (int i = 0; i < 10; i++) {
            try {
                return dao.packByPackageNumber_readHistory( packageNumber );                
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                logMemory("系统发生异常", e.getMessage());
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e1) {
                    // TODO Auto-generated catch block
                    e1.printStackTrace();
                }
            } 
        }// loop end
        return null ;
        
    }
    
    
    /**
     * 读 著录数据
     * @return
     */
    private String packByPackageNumber_readBibliographic( String patentId ){

        StringBuffer sb = new StringBuffer();
        
        try {
            pbcList = dao.packByPackageNumber_readBibliographic( patentId );
            
            for( PatentBibliographicChangeBean pbcBean : pbcList ){
                // 读取 著录 数据
                sb.append( pbcBean.toString() );
            }
            
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } 
        return sb.toString() ;
    }
    
    private JSONObject packByPackageNumber_readBibl( String patentId ){

        StringBuffer sb = new StringBuffer();
        JSONObject json = new JSONObject();        

        for (int i = 0; i < 10; i++) {
            try {
                pbcList = dao.packByPackageNumber_readBibliographic( patentId );
                
                for( PatentBibliographicChangeBean pbcBean : pbcList ){
                    // 读取 著录 数据
                    json = JSONObject.fromObject( pbcBean );
                }
                return json ;
                
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                logMemory("系统发生异常", e.getMessage());
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e1) {
                    // TODO Auto-generated catch block
                    e1.printStackTrace();
                }
            } 
        }// loop end
        
        return null ;
    }
    

    /**
     * 插入 MongoDB
     */
    public void insertMongoDB( List<PatentBibliographicChangeBean> pbcList ) {    

        //# MongoDB(数据加载目标)
        String syncLoadIntoMongoDbService = "10.78.2.23:27017";
        String syncLoadIntoMongoDbName = "patent_search_extend";
        String syncLoadIntoMongoTable = "patent_bibliographic_20160319";

        // 加载开始
//        logger.info(DateUtils.getNow() + " Load start: " + syncLoadIntoMongoDbService );

        Mongo m = null;
        try {
            m = new Mongo( syncLoadIntoMongoDbService );
        } catch (UnknownHostException e) {
            e.printStackTrace();
            logger.info(DateUtils.getNow() + " UnknownHostException:" + e.getMessage());
        }

        // 库名
        DB db = m.getDB( syncLoadIntoMongoDbName );
//        logger.info( DateUtils.getNow() + " Db:" + syncLoadIntoMongoDbName );

        // 表名
        DBCollection collection = db.getCollection( syncLoadIntoMongoTable );
//        logger.info(DateUtils.getNow() + " Table:" + syncLoadIntoMongoTable );
        
        // 循环列表，将每个元素插入数据库
        for( PatentBibliographicChangeBean pbcBean : pbcList ){        
            
            // 读取一条著录数据
            JSONObject json = packByPackageNumber_readBibl( pbcBean.getId() );
            
            if( null == json ) return ;

            // 列，值
            BasicDBObject insDoc = new BasicDBObject();
            
            insDoc.put("abstract_No"    , json.get( "abstract_No"    ).toString());
            insDoc.put("app_Addr"       , json.get( "app_Addr"       ).toString());
            insDoc.put("app_Cn"         , json.get( "app_Cn"         ).toString());
            insDoc.put("app_Country"    , json.get( "app_Country"    ).toString());
            insDoc.put("app_Date"       , json.get( "app_Date"       ).toString());
            insDoc.put("app_Name"       , json.get( "app_Name"       ).toString());
            insDoc.put("app_Sn"         , json.get( "app_Sn"         ).toString());
            insDoc.put("app_Type"       , json.get( "app_Type"       ).toString());
            insDoc.put("app_Zip"        , json.get( "app_Zip"        ).toString());
            insDoc.put("ecla"           , json.get( "ecla"           ).toString());
            insDoc.put("fi"             , json.get( "fi"             ).toString());
            insDoc.put("ft"             , json.get( "ft"             ).toString());
            insDoc.put("id"             , json.get( "id"             ).toString());
            insDoc.put("inv_Title"      , json.get( "inv_Title"      ).toString());
            insDoc.put("invent_Type"    , json.get( "invent_Type"    ).toString());
            insDoc.put("inventor"       , json.get( "inventor"       ).toString());
            insDoc.put("ipc_Standard"   , json.get( "ipc_Standard"   ).toString());
            insDoc.put("locarno"        , json.get( "locarno"        ).toString());
            insDoc.put("operation_Time" , json.get( "operation_Time" ).toString());
            insDoc.put("operation_Type" , json.get( "operation_Type" ).toString());
            insDoc.put("package_Name"   , json.get( "package_Name"   ).toString());
            insDoc.put("pct_App_Cn"     , json.get( "pct_App_Cn"     ).toString());
            insDoc.put("pct_App_Date"   , json.get( "pct_App_Date"   ).toString());
            insDoc.put("pct_App_Sn"     , json.get( "pct_App_Sn"     ).toString());
            insDoc.put("pct_Date"       , json.get( "pct_Date"       ).toString());
            insDoc.put("pct_Pub_Cn"     , json.get( "pct_Pub_Cn"     ).toString());
            insDoc.put("pct_Pub_Date"   , json.get( "pct_Pub_Date"   ).toString());
            insDoc.put("pct_Pub_Lang"   , json.get( "pct_Pub_Lang"   ).toString());
            insDoc.put("pct_Pub_Sn"     , json.get( "pct_Pub_Sn"     ).toString());
            insDoc.put("prn"            , json.get( "prn"            ).toString());
            insDoc.put("prn_Cn"         , json.get( "prn_Cn"         ).toString());
            insDoc.put("prn_Date"       , json.get( "prn_Date"       ).toString());
            insDoc.put("prn_Sn"         , json.get( "prn_Sn"         ).toString());
            insDoc.put("prn_Type"       , json.get( "prn_Type"       ).toString());
            insDoc.put("pub_Cn"         , json.get( "pub_Cn"         ).toString());
            insDoc.put("pub_Date"       , json.get( "pub_Date"       ).toString());
            insDoc.put("pub_Sn"         , json.get( "pub_Sn"         ).toString());
            insDoc.put("pub_Type"       , json.get( "pub_Type"       ).toString());
            insDoc.put("uc"             , json.get( "uc"             ).toString());
            
            collection.insert(insDoc);
            insDoc.clear();
            
        }

        // 循环遍历pdaBeanList
//        System.out.println("loading ...");
//        logger.info(DateUtils.getNow() + " rows:" + pdaBeanList.size());

        // 当前记录编号
        // int currentRowNumber = 0;

        if ( m != null) m.close();
        
//        System.out.println("Load finished.");
    }
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    

    /** 记录日志 */
    private void logMemory(String behavior, String content) {
        // 向服务器发送日志
//        sendLog.send("syncPacker", behavior, content);
        // 记录本地日志
        logger.info(DateUtils.getNow() + " " + behavior + " :" + content);
        // 控制台输出日志
//        System.out.println("syncPacker : " + DateUtils.getNow() + " " + behavior + " :" + content);
    }

    @Override
    public String insert() throws Exception {
        return null;
    }

    @Override
    public String update() throws Exception {
        return null;
    }

    @Override
    public String selectList() throws Exception {
        return null;
    }

    @Override
    public String delete() throws Exception {
        return null;
    }

    public static boolean isRunning() {
        return isRunning;
    }

    public static void setRunning(boolean isRunning) {
        SyncPacker_201603.isRunning = isRunning;
    }

    public Integer getCenterNodeDataVersion() {
        return centerNodeDataVersion;
    }

    public void setCenterNodeDataVersion(Integer centerNodeDataVersion) {
        this.centerNodeDataVersion = centerNodeDataVersion;
    }

    public String getSuccess() {
        return success;
    }

    public void setSuccess(String success) {
        this.success = success;
    }

    public String getMaxPackageNumber() {
        return maxPackageNumber;
    }

    public void setMaxPackageNumber(String maxPackageNumber) {
        this.maxPackageNumber = maxPackageNumber;
    }

    public String getPackedPackageNumber() {
        return packedPackageNumber;
    }

    public void setPackedPackageNumber(String packedPackageNumber) {
        this.packedPackageNumber = packedPackageNumber;
    }

    public SyncDataPackageBean getBean() {
        return bean;
    }

    public void setBean(SyncDataPackageBean bean) {
        this.bean = bean;
    }
}

# log4j.rootLogger = INFO, stdout

log4j.rootLogger= INFO, syncLoader

# + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - +
# 数据加载器（ syncLoader ）
# + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - +
log4j.logger.com.neusoft.patent.search.extend.syncLoader.SyncLoaderAction = INFO, syncLoader
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - 
log4j.appender.syncLoader.File           = ${webapp.root}/logs/syncLoader_20160318/log.html
log4j.appender.syncLoader                = org.apache.log4j.RollingFileAppender
log4j.appender.syncLoader.Encoding       = GBK
log4j.appender.syncLoader.MaxFileSize    = 5MB
log4j.appender.syncLoader.MaxBackupIndex = 100
log4j.appender.syncLoader.Threshold      = INFO
log4j.appender.syncLoader.layout         = org.apache.log4j.HTMLLayout

SyncPackerDao.xml

<?xml version="1.0" encoding="UTF-8" ?> 
<!DOCTYPE mapper 
    PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" 
    "http://mybatis.org/dtd/mybatis-3-mapper.dtd">

<mapper namespace="syncPacker.SyncPackerDao">

    <!-- 查询分块著录信息 -->
    <select id="selectList"
        parameterType="syncPacker.bean.SyncDataPackageBean"
        resultType="syncPacker.bean.PatentBibliographicChangeBean">
            select t.id,
                   t.operation_time,
                   t.operation_type,
                   sv.app_cn,
                   sv.app_sn,
                   sv.app_type,
                   sv.app_date,
                   sv.pub_cn,
                   sv.pub_sn,
                   sv.pub_type,
                   sv.pub_date,
                   sv.invent_type,
                   sv.app_name,
                   sv.inventor,
                   sv.app_country,
                   sv.app_addr,
                   sv.app_zip,
                   sv.inv_title,
                   sv.abstract_no,
                   sv.pct_date,
                   sv.pct_app_cn,
                   sv.pct_app_sn,
                   sv.pct_app_date,
                   sv.pct_pub_cn,
                   sv.pct_pub_sn,
                   sv.pct_pub_lang,
                   sv.pct_pub_date,
                   sv.ipc_standard,
                   sv.ecla,
                   sv.uc,
                   sv.fi,
                   sv.ft,
                   sv.prn_cn,
                   sv.prn_sn,
                   sv.prn_type,
                   sv.prn_date,
                   sv.prn,
                   sv.locarno
              from (select t.id, t.operation_time, t.operation_type
                     from ${tableName} t
                     where rownum &lt;= ${maxRowsPerSyncPackerPackage}
                     order by operation_time asc) t
              left join sipo_vdb sv
                on t.id = sv.id
    </select>
         
    <!-- 添加数据至归档表 -->
    <insert id="appendIntoChangeHistory"
        parameterType="syncPacker.bean.PatentBibliographicChangeBean">
            insert into e_bibliographic_change_history
              (id,
               operation_time,
               operation_type,
               package_number)
            values
              (#{Id},
               #{Operation_Time},
               #{Operation_Type},
               #{Package_Name})
    </insert>

    <!-- 删除操作表的数据 -->
    <delete id="removeFromChangeTable" parameterType="java.util.List">
        delete from
            e_bibliographic_change_temp 
        where id in
        <foreach collection="list" index="index" item="item" open="(" separator="," close=")">
            #{item}
        </foreach>
    </delete>

    <!-- 获取当前最大序列号的数据包信息 -->
    <select id="getMaxPackageNumber" resultType="String">
        select
            ifnull(max(sync_data_package.packageNumber),0) packageNumber
        from
            sync_data_package
    </select>

    <!-- 添加数据包编号至发送记录表 -->
    <insert id="insertSyncDataPackage" parameterType="String">
        insert into
            sync_data_package
            (packageNumber)
        values
            (#{theMaxSerialNumber})
    </insert>
    
    <!-- 重新打包指定数据 -->
    <select id="repackList"
        parameterType="syncPacker.bean.SyncDataPackageBean"
        resultType="syncPacker.bean.PatentBibliographicChangeBean">
        select t.id,
               t.operation_time,
               t.operation_type,
               sv.app_cn,
               sv.app_sn,
               sv.app_type,
               sv.app_date,
               sv.pub_cn,
               sv.pub_sn,
               sv.pub_type,
               sv.pub_date,
               sv.invent_type,
               sv.app_name,
               sv.inventor,
               sv.app_country,
               sv.app_addr,
               sv.app_zip,
               sv.inv_title,
               sv.abstract_no,
               sv.pct_date,
               sv.pct_app_cn,
               sv.pct_app_sn,
               sv.pct_app_date,
               sv.pct_pub_cn,
               sv.pct_pub_sn,
               sv.pct_pub_lang,
               sv.pct_pub_date,
               sv.ipc_standard,
               sv.ecla,
               sv.uc,
               sv.fi,
               sv.ft,
               sv.prn_cn,
               sv.prn_sn,
               sv.prn_type,
               sv.prn_date,
               sv.prn,
               sv.locarno
          from (select t.id, t.operation_time, t.operation_type, t.package_number
                  from E_BIBLIOGRAPHIC_CHANGE_HISTORY t
                 where t.package_number = #{packageNumber}
                 order by transfer_time asc) t
          left join sipo_vdb sv
            on t.id = sv.id
    </select>
    
    
    <!-- 
        重新打包: 读取著录变化数据
        参数：包编号
     -->        
    <select id="selectList_changeHistory"
        parameterType="java.lang.String"
        resultType="syncPacker.bean.PatentBibliographicChangeBean">
        
        select            
            sv.id,
            sv.app_cn,
            sv.app_sn,
            sv.app_type,
            sv.app_date,
            sv.pub_cn,
            sv.pub_sn,
            sv.pub_type,
            sv.pub_date,
            sv.invent_type,
            sv.app_name,
            sv.inventor,
            sv.app_country,
            sv.app_addr,
            sv.app_zip,
            sv.inv_title,
            sv.abstract_no,
            sv.pct_date,
            sv.pct_app_cn,
            sv.pct_app_sn,
            sv.pct_app_date,
            sv.pct_pub_cn,
            sv.pct_pub_sn,
            sv.pct_pub_lang,
            sv.pct_pub_date,
            sv.ipc_standard,
            sv.ecla,
            sv.uc,
            sv.fi,
            sv.ft,
            sv.prn_cn,
            sv.prn_sn,
            sv.prn_type,
            sv.prn_date,
            sv.prn,
            sv.locarno
            
        from 
            sipo_vdb sv
            
        where 
            exists ( 
                select 1 from e_bibliographic_change_history t 
                where sv.ID = t.id and t.package_number = #{packageNumber}  
            )
            
    </select>
    
</mapper>