最近在做数据批量导入的工作（excel批量导入数据库软件）

场景：从ORACLE中读取5K万++的大表，读取到指定条件数据到MYSQL数据库中。

ORACLE库为第三方公司的，只提供了DBlink方式连接，查询速度很慢。

刚开始理想了，想按指定日期查询出数据再切分list，再批量保存到库中。查询一天就60多W条，执行速度太慢，等待时间长，容易超时。

后来调整为根据指定条件查询总记录数，再分页查出记录0-1000条，再通过10个线程去跑1000条记录保存入库。

过程一、

SELECT count(1)  FROM  emp t  WHERE hire_date BETWEEN TO_DATE ('20060501', 'yyyymmdd')

                                       AND TO_DATE ('20060731', 'yyyymmdd')

过程二、

SELECT  *

  FROM (SELECT tt.*, ROWNUM AS rowno

          FROM (  SELECT t.*

                    FROM emp t

                   WHERE hire_date BETWEEN TO_DATE ('20060501', 'yyyymmdd')

                                       AND TO_DATE ('20060731', 'yyyymmdd')

                ORDER BY create_time DESC, emp_no) tt

         WHERE ROWNUM <= 20) table_alias

 WHERE table_alias.rowno >= 10 

过程三、

private void toDo(String startYmd, String endYmd,int count) throws InterruptedException {


        log.info("当前时间内，开始时间：{} 结束时间：{} 类型：'{}' 总记录数为：{} ",startYmd,endYmd,type,count);

        int rownumStart = 0;
        int rownumEnd = 0;
        int len = 2000;
        int remainder = count%len ; // 余数
        int size = count/len;
        for (int i = 0; i < size; i++) {
            rownumStart = i*len;
            rownumEnd = (i+1)*len;
            if(i==size-1){
                rownumEnd = rownumEnd+remainder;
            }
                executeQuery(startYmd, endYmd, rownumStart, rownumEnd);
        }
    }

/**
     * 通过分页查询，把数据量减小，以完成数据检索。
     * @param startYmd
     * @param endYmd
     * @param rownumStart
     * @param rownumEnd
     * @return
     * @throws InterruptedException
     */
    private void executeQuery(String startYmd, String endYmd, int rownumStart, int rownumEnd) throws InterruptedException {

        log.info("分页查询参数为：rownumStart:{} rownumEnd:{} ",rownumStart,rownumEnd);

        /**
         * 从数据源中读取数据
         */
        List<CustHis96Lc> allList = custHis96lcMapper2.queryAll(startYmd,endYmd,rownumStart,rownumEnd);

        int dataSize = allList.size();

        log.info("》》》》》》》》》》》》》》当前rownum下 客户负荷信息采集 数据总共{}条.....",dataSize);

        int runSize = 10; // 开启的线程数
        if (allList!=null && dataSize>0){
            int size = allList.size();
            int length = dataSize/runSize;
            execute(allList,runSize,length);
        }
    }

private void execute(List<CustHis96Lc> allList, int runSize, int length) {

    log.info("运行线程数据 ：{} 运行数据长度：{}",runSize,length);
    /**
     * 到这里进行相关逻辑的处理 如行业类型匹配、单位对照
     * 最后进行保存.
     */
    int listSize = allList.size(); // 集合大小
    ExecutorService executor = Executors.newFixedThreadPool(runSize);

    // 创建两个个计数器
    List<CustHis96Lc> newList = null;
    for (int i = 0; i < runSize; i++) {
        if ((i + 1) == runSize) {
            int startIdx = (i * length);
            int endIdx = allList.size();
            newList = allList.subList(startIdx, endIdx);
        } else {
            int startIdx = (i * length);
            int endIdx = (i + 1) * length;
            newList = allList.subList(startIdx, endIdx);
        }
        // 多线程处理
        CustHisLcThreadHandle thread = new CustHisLcThreadHandle(newList, this.custHis96lcMapper1);
        executor.execute(thread);
    }
    executor.shutdown();
}

/**
 * 多线程处理台区历史负荷数据
 */
@Slf4j
public class CustHisLcThreadHandle implements Runnable {

    private static Semaphore s = new Semaphore(10);

    /**
     * 数据集合
     */
    List<CustHis96Lc> list;

    private CustHis96lcMapper1 custHis96lcMapper1;

    public CustHisLcThreadHandle(List<CustHis96Lc> list,
                                 CustHis96lcMapper1 custHis96lcMapper1) {
        this.list = list;
        this.custHis96lcMapper1 = custHis96lcMapper1;
    }


    @Override
    public void run() {
        log.info("开始线程>>>>>>>>>>>");
        try {
                if (list != null && !list.isEmpty()) {
                    // 执行操作
                    s.acquire();
                    List<List<CustHis96Lc>> rsList = Lists.partition(list, 100);
                    rsList.forEach(item -> custHis96lcMapper1.saveOrUpdate(item));
                    s.release();
                }
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            if (s!=null){
                s.release();
            }
        }
    }
}

插入操作采用Mybatis操作

<insert id="saveOrUpdate">
        insert into LF_CTRL_TAIQU_INFO(TAIQUID,TAIQUNAME, NETID, tradeID, capacity, taiqucode)
        VALUES
        <foreach collection="list" item="item" separator=",">
            (#{item.taiquid},#{item.taiquname},#{item.netid},#{item.tradeid},#{item.capacity},#{item.taiqucode})
        </foreach>
        ON DUPLICATE KEY UPDATE
        NETID=values(NETID),
        tradeID=values(tradeID),
        capacity=values(capacity),
        taiqucode=values(taiqucode)
    </insert>

这样就完成了批量插入操作。

但效果还不是很好，因为插入目标数据库有IO瓶颈，后期考虑多库插入，再库库同步，应该会快很多。