多线程分页处理批量数据--jdbc方式

/**
     * 同步数据信息到ES
     * @return
     */
    public boolean syncNhReportSeessToEs(){

        long begin = System.currentTimeMillis();
        logger.error("------------------开始始初始化报备带看列表数据到ES"+ DateUtil.getDefaultDate()+ "-----------------------");
        //分页大小
        int pageSize = 500;
            int total = jdbcTemplate.queryForInt("select count(1) from (select * from nh_report nr where nr.id in(select min(nr1.id) from nh_report nr1 where nr1.record_is_deleted = 0 and nr1.city_code = 1 group by nr1.reporter_panid,nr1.customer_name,nr1.customer_phone))");
            if (total == 0) {
                return true;
            }
            logger.error("查出需要处理的报备表数据为"+total+"条");
            double querySize = total;
        double totalPage = Math.ceil(querySize / pageSize);
        logger.error("共分" + totalPage + "页");
        final String sql = "select * from(" +
                "select nr.*,rownum as rn from nh_report nr where nr.id in" +
                "(select min(nr1.id) from nh_report nr1" +
                " where nr1.record_is_deleted = 0 and nr1.city_code = 1 group by nr1.reporter_panid, nr1.customer_name, nr1.customer_phone) and rownum <= ? ）where rn>=?";
        ExecutorService es = Executors.newFixedThreadPool(20);
            for (int i = 0; i < totalPage; i++) {
                logger.error("执行第"+i+"页");
                int start = i * pageSize + 1;
                int end = (i + 1) * pageSize;
                if (end > total) {
                    end = total;
                }
                final Object[] params = new Object[] { end, start };
                es.submit(new Runnable() {
                    @SuppressWarnings("unused")
                    @Override
                    public void run() {
                        List<Map<String, Object>> objs = new ArrayList<Map<String, Object>>();
                        List<ReportSeeEsModel> contants = jdbcTemplate.query(sql, params, new RowMapper() {
                            @Override
                            public ReportSeeEsModel mapRow(ResultSet rs, int i) throws SQLException {
                                ReportSeeEsModel model = new ReportSeeEsModel();
                                String id = JugHelper.generalUUID();
                                //主键id
                                model.setId(id);
                                model.setCityCode(String.valueOf(NhReport.CITYCODE_BEIJING));
                                //客户姓名
                                String customerName = rs.getObject("CUSTOMER_NAME") != null ? rs.getObject("CUSTOMER_NAME").toString() : "";
                                model.setCustomerName(customerName);
                                //客户姓名 不分词
                                model.setReporterCustomerName(customerName);
                                //客户电话
                                String customerPhone = rs.getObject("CUSTOMER_PHONE") != null ? rs.getObject("CUSTOMER_PHONE").toString() : "";
                                model.setCustomerPhone(customerPhone);
                                //客户电话，不分词
                                model.setReporterCustomerPhone(customerPhone);
                                //报备人公盘id
                                String reportPanId = rs.getObject("REPORTER_PANID") != null ? rs.getObject("REPORTER_PANID").toString() : "";
                                model.setReporterPanid(reportPanId);
                                //城市编码
                                model.setCityCode(String.valueOf(NhReport.CITYCODE_BEIJING));
                                //部门id
                                model.setDeptId(rs.getObject("DEPT_ID") != null ? rs.getObject("DEPT_ID").toString() : "");
                                //店组id
                                model.setShopId(rs.getObject("SHOP_ID") != null ? rs.getObject("SHOP_ID").toString() : "");
                                //大区
                                model.setRegionId(rs.getObject("REGION_ID") != null ? rs.getObject("REGION_ID").toString() : "");
                                return model;
                            }
                        });

                        //循环处理带看和户型大小的相关数据
                        for (ReportSeeEsModel reportSeeEsModel:contants
                             ) {
                            String reporterPanid = reportSeeEsModel.getReporterPanid();
                            String customerName = reportSeeEsModel.getCustomerName();
                            String customerPhone = reportSeeEsModel.getCustomerPhone();
                            final Object[] params = new Object[] { reporterPanid,customerName,customerPhone };
                            //查询当前经纪人，当前客户姓名，当前客户电话相关的带看
                            String sql = "select * from nh_report_see nrs" +
                                    " join nh_report nr on nr.id = nrs.report_id" +
                                    " where nr.reporter_panid = ? and nr.customer_name = ? and nr.customer_phone = ? and nrs.see_time is not null";
                            List<NhReportSee> reportSees = jdbcTemplate.query(sql, params, new RowMapper() {
                                @Override
                                public NhReportSee mapRow(ResultSet rs, int i) throws SQLException {
                                    NhReportSee reportSee = new NhReportSee();
                                    reportSee.setReportId(rs.getObject("REPORT_ID") != null ? rs.getObject("REPORT_ID").toString() : "");
                                    reportSee.setSeeTime(rs.getObject("SEE_TIME") != null ? (Date) rs.getObject("SEE_TIME") : null);
                                    reportSee.setId(rs.getObject("ID") != null ? rs.getObject("ID").toString() : "");
                                    reportSee.setStageId(rs.getObject("STAGE_ID") != null ? rs.getObject("STAGE_ID").toString() : "");
                                    return reportSee;
                                }
                            });
                            //根据当前经纪人当前客户的所有报备id查询出所有的意向户型
                            String sql2 = "select * from nh_stage_layout nsl" +
                                    " join nh_report_layout nrl on nsl.id = nrl.layout_id" +
                                    " join nh_report nr on nr.id  = nrl.proposer_id" +
                                    " where nr.reporter_panid = ? and nr.customer_name = ? and nr.customer_phone = ?";
                            List<LayoutVo> layouts = jdbcTemplate.query(sql2, params, new RowMapper() {
                                @Override
                                public LayoutVo mapRow(ResultSet rs, int i) throws SQLException {
                                    LayoutVo layoutVo = new LayoutVo();
                                    layoutVo.setArea(rs.getObject("AREA") != null ? rs.getObject("AREA").toString() : "");
                                    layoutVo.setRoomNum(rs.getObject("ROOM_NUM") != null ? rs.getObject("ROOM_NUM").toString() : "");
                                    layoutVo.setTotalPrice(rs.getObject("TOTAL_PRICE") != null ? Integer.valueOf(rs.getObject("TOTAL_PRICE").toString()) : null);
                                    return layoutVo;
                                }
                            });
                            //查询当前经纪人当前客户的所有报备
                            String sql3 = "select * from nh_report nr where nr.reporter_panid = ? and nr.customer_name = ? and nr.customer_phone = ? order by nr.create_date desc";
                            List<NhReport> reports = jdbcTemplate.query(sql3, params, new RowMapper() {
                                @Override
                                public NhReport mapRow(ResultSet rs, int i) throws SQLException {
                                    NhReport report = new NhReport();
                                    report.setId(rs.getObject("ID") != null ? rs.getObject("ID").toString() : "");
                                    report.setCreateDate(rs.getObject("CREATE_DATE") != null ? (Date) rs.getObject("CREATE_DATE") : null);
                                    return report;
                                }
                            });
                            //报备时间
                            reportSeeEsModel.setReportTime(reports.get(0).getCreateDate().toString());
                            String areaMax = CommonUtil.formatDouble(CommonUtil.getMaxLayoutArea(layouts)).equals("") ? "0" : CommonUtil.formatDouble(CommonUtil.getMaxLayoutArea(layouts));
                            String areaMin = CommonUtil.formatDouble(CommonUtil.getMinLayoutArea(layouts)).equals("") ? "0" : CommonUtil.formatDouble(CommonUtil.getMinLayoutArea(layouts));
                            String roomMax = CommonUtil.formatDouble(CommonUtil.getMaxLayoutHou(layouts)).equals("") ? "0" : CommonUtil.formatDouble(CommonUtil.getMaxLayoutHou(layouts));
                            String roomMin = CommonUtil.formatDouble(CommonUtil.getMinLayoutHou(layouts)).equals("") ? "0" : CommonUtil.formatDouble(CommonUtil.getMinLayoutHou(layouts));
                            String priceMax = CommonUtil.formatDouble(CommonUtil.getMaxLayoutTotalPrice(layouts)).equals("") ? "0" : CommonUtil.formatDouble(CommonUtil.getMaxLayoutTotalPrice(layouts));
                            String priceMin = CommonUtil.formatDouble(CommonUtil.getMinLayoutTotalPrice(layouts)).equals("") ? "0" : CommonUtil.formatDouble(CommonUtil.getMinLayoutTotalPrice(layouts));
                            //最大价格
                            reportSeeEsModel.setHouPriceMax(String.valueOf(priceMax));
                            //最小价格
                            reportSeeEsModel.setHouPriceMin(String.valueOf(priceMin));
                            //最大面积
                            reportSeeEsModel.setHouAreaMax(String.valueOf(areaMax));
                            //最小面积
                            reportSeeEsModel.setHouAreaMin(String.valueOf(areaMin));
                            //最大居室
                            reportSeeEsModel.setHouMax(String.valueOf(String.valueOf(roomMax)));
                            //最小居室
                            reportSeeEsModel.setHouMin(String.valueOf(String.valueOf(roomMin)));
                            //带看次数
                            reportSeeEsModel.setSeeCount(String.valueOf(reportSees.size()));
                            Set<NhReportSee> set = new TreeSet<NhReportSee>(new Comparator<NhReportSee>() {
                                @Override
                                public int compare(NhReportSee a, NhReportSee b) {
                                    // 字符串则按照asicc码升序排序
                                    return a.getStageId().compareTo(b.getStageId());
                                }
                            });
                            set.addAll(reportSees);
                            List<NhReportSee> list = new ArrayList<>(set);
                            List<ReportSeeInfoEsModel> seeinfos = new ArrayList<>();
                            if(!CollectionUtils.isEmpty(list)){//如果带看的条数不为0
                                int index = 0;
                                for (NhReportSee see:list
                                ) {
                                    if(index==2){
                                        break;
                                    }
                                    ReportSeeInfoEsModel reportSeeInfoEsModel = new ReportSeeInfoEsModel();

                                    final Object[] params1 = new Object[] { see.getReportId() };
                                    String sql4 = "select * from nh_report nr where nr.id = ?";
                                    NhReport nhReport = (NhReport) jdbcTemplate.queryForObject(sql4, params1, new RowMapper() {
                                        @Override
                                        public NhReport mapRow(ResultSet rs, int i) throws SQLException {
                                            NhReport nhReport = new NhReport();
                                            nhReport.setId(rs.getObject("ID") != null ? rs.getObject("ID").toString() : "");
                                            nhReport.setIsCustomerA(rs.getObject("IS_CUSTOMER_A") != null ? Integer.valueOf(rs.getObject("IS_CUSTOMER_A").toString()) : null);
                                            nhReport.setIsFirstVisit(rs.getObject("IS_FIRST_VISIT") != null ? Integer.valueOf(rs.getObject("IS_FIRST_VISIT").toString())  : null);
                                            nhReport.setIsEffective(rs.getObject("IS_EFFECTIVE") != null ? Integer.valueOf(rs.getObject("IS_EFFECTIVE").toString())  : null);
                                            nhReport.setAuditStatus(rs.getObject("AUDIT_STATUS") != null ? Integer.valueOf(rs.getObject("AUDIT_STATUS").toString())  : null);
                                            nhReport.setIntentionStageId(rs.getObject("INTENTION_STAGE_ID") != null ? rs.getObject("INTENTION_STAGE_ID").toString() : "");
                                            nhReport.setPerformanceId(rs.getObject("PERFORMANCE_ID") != null ? rs.getObject("PERFORMANCE_ID").toString() : "");
                                            return nhReport;
                                        }
                                    });

                                    //是否是A类
                                    reportSeeInfoEsModel.setIsCustomerA(String.valueOf(nhReport.getIsCustomerA()));
                                    //是否是首访
                                    reportSeeInfoEsModel.setIsFirstVisit(String.valueOf(nhReport.getIsFirstVisit()));
                                    //报备id
                                    reportSeeInfoEsModel.setReportId(nhReport.getId());
                                    //带看时间
                                    reportSeeInfoEsModel.setSeeDate(see.getSeeTime());
                                    //认购时间和草签时间
                                    if(StringUtils.isNotBlank(nhReport.getPerformanceId())){
                                        final Object[] params2 = new Object[] { see.getReportId() };
                                        String sql5 = "select * from nh_sale_performance nsp where nsp.report_id = ?";
                                        NhSalePerformance performance = (NhSalePerformance) jdbcTemplate.queryForObject(sql5, params2, new RowMapper() {
                                            @Override
                                            public NhSalePerformance mapRow(ResultSet rs, int i) throws SQLException {
                                                NhSalePerformance performance = new NhSalePerformance();
                                                performance.setId(rs.getObject("ID") != null ? rs.getObject("ID").toString() : "");
                                                performance.setBuyDate(rs.getObject("BUY_DATE") != null ? (Date)rs.getObject("BUY_DATE")  : null);
                                                performance.setDraftDate(rs.getObject("DRAFT_DATE") != null ? (Date)rs.getObject("DRAFT_DATE")  : null);
                                                return performance;
                                            }
                                        });
                                        if(performance.getBuyDate()!=null){
                                            reportSeeInfoEsModel.setBuyDate(performance.getBuyDate());
                                        }
                                        if(performance.getDraftDate()!=null){
                                            reportSeeInfoEsModel.setDraftDate(performance.getDraftDate());
                                        }
                                    }
                                    //带看id
                                    reportSeeInfoEsModel.setSeeId(see.getId());
                                    //是否有效
                                    reportSeeInfoEsModel.setIsEffective(String.valueOf(nhReport.getIsEffective()));
                                    //审核状态
                                    reportSeeInfoEsModel.setAuditStatus(String.valueOf(nhReport.getAuditStatus()));
                                    //楼盘id
                                    reportSeeInfoEsModel.setStageId(nhReport.getIntentionStageId());
                                    NhStage stage = nhStageIService.findById(nhReport.getIntentionStageId());
                                    if(stage!=null){
                                        //楼盘名称
                                        reportSeeInfoEsModel.setStageName(stage.getStageName());
                                    }
                                    seeinfos.add(reportSeeInfoEsModel);
                                    index++;
                                }
                            }
                            reportSeeEsModel.setSeeInfo(seeinfos);
                        }
                        for (ReportSeeEsModel model : contants) {
                            try {
                                objs.add(objectToMap(model));
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                        if(objs.size()>0){
                            logger.error("本次插入数据"+objs.size()+"条");
                            String result = ESClient.batchInsert(ESContant.NEWHOUSECUSTOMER_INDEX, ESContant.REPORTCUSTOMER_TYPE, objs);
                            logger.error(result);
                        }

                    }
                });
            }
            es.shutdown();
            try {
                boolean loop = true;
                do {    //等待所有任务完成
                    loop = !es.awaitTermination(2, TimeUnit.SECONDS);  //阻塞，直到线程池里所有任务结束
                } while (loop);
            } catch (Exception e) {
                e.printStackTrace();
            }
            logger.error("------------------结束初始化报备带看列表数据到ES"+ DateUtil.getDefaultDate()+ "-----------------------");
            return true;
    }