多线程分页处理批量数据--jdbc方式
2024-09-01 13:13:39
/**
* 同步数据信息到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;
}
最新文章
- longjmp setjmp and volatile
- MongoDB-集群搭建
- Yii2美化confirm
- ubuntu 13.04 lighttped mysql php
- hdu1331 按着题目的公式直接写
- hadoop2.2编程：MRUnit
- unity创建和加载AssetBundle
- MongoDB 原子操作
- Centos7配置hadoop伪分布式
- python 三目运算符
- smartctl 检测磁盘信息
- RPM 包的构建 - 实例
- 在云端服务器centos7安装jvm并且运行java程序
- Win10系列：VC++绘制几何图形3
- @RequestMapping、@Responsebody、@RequestBody和@PathVariable详解（转）
- SqlServer安装时的选项说明
- elasticsearch 多列 聚合（sql group by）
- 解决错误：Couldn't open file /etc/pki/rpm-gpg/RPM-GPG-KEY-EPEL-7
- python3的一些练习题
- SSIS实践入门3：把SSIS程序包从A电脑到B电脑的转移