导出百万级-EasyExcel实践二

朴灿烈づ我的快乐病毒、 2023-06-27 08:45 62阅读 0赞

1：多线程高并发[【java并发核心一】Semaphore 的使用思路][java_Semaphore]

2：高并发下，线程安全

2s 10w数据量导出，待优化

int ndrcNum = SyuserSyroleMapper.countUserIsNDRC(dto.getSyuserId());
            Query sql= getMatchFieldsByQuery(dto, bo,ndrcNum);
            //计算总数量，进行分批
            long totalRowCount = mongoTemplate.count(sql,"notificationChargeOrderInfo");
                System.out.println("导出总数totalRowCount|"+totalRowCount);
            List<Future> headerflist = new ArrayList<>();
            Long toIndex = Consts.PER_SHEET_ROW_COUNT_10W.longValue(); //数据100000条处理存一次
                int sheetIndex=0;//根据分批数创建对用数sheet
    
                // 这里 指定文件
    //         WriteSheet writeSheet = EasyExcel.writerSheet("sheetName0").build();
    //         Map<Integer,Object> sheetMap=new HashMap<Integer,Object>();
                // 去调用写入,这里我调用了五次，实际使用时根据数据库分页的总的页数来。这里最终会写到5个sheet里面
                List<List<ExportOrder2Bo>> orderLists=new ArrayList<List<ExportOrder2Bo>>();
                for (int i = 0; i < totalRowCount; i += Consts.PER_SHEET_ROW_COUNT_10W) {
    
                    //主线程休息等待从thread把上一个线程sql条件处理完毕（重点一）
                    if (i != 0) {
                        Thread.sleep(50);
                    }
                    if (i + Consts.PER_SHEET_ROW_COUNT_10W > totalRowCount) {        //作用为toIndex最后没有100条数据则剩余几条newList中就装几条
                    toIndex = totalRowCount - i;
                    }
                    // 并发执行
                    int startIndex = i;//开始位置
                    int limitSizes=toIndex.intValue();//分页条数
                sql.skip(startIndex).limit(limitSizes);
    //            logger.info(Thread.currentThread().getName() + "|11111111mongo sql"+JSON.toJSON(sql));
                System.out.println(Thread.currentThread().getName()+"|11111111Runnable：startIndex:"+startIndex+"|limitSizes:"+limitSizes);
                    // 每次都要创建writeSheet 这里注意必须指定sheetNo
    //            System.out.println("writeSheet开启sheetIndex："+sheetIndex);
    //            final String sheetName="sheet"+sheetIndex;
    //            WriteSheet writeSheet = EasyExcel.writerSheet(sheetIndex,sheetName).build();
    //            sheetMap.put(sheetIndex,writeSheet);
    //            System.out.println("第"+sheetIndex+"批|sheetName:"+writeSheet.getSheetName()+"|sheetIndex: "+sheetIndex);
                headerflist.add(processExportOrderExcelPool.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            /**
                             * 在 semaphore.acquire() 和 semaphore.release()之间的代码，同一时刻只允许制定个数的线程进入，
                             * 因为semaphore的构造方法是1，则同一时刻只允许一个线程进入，其他线程只能等待。
                             * */
    //                        //根据条件查询订单（并行处理）
                            System.out.println(Thread.currentThread().getName()+"|22222222Runnable：startIndex:"+startIndex+"|limitSizes:"+limitSizes+"|alexseq|"+JSON.toJSON(sql));
                            long tt2=System.currentTimeMillis();
                            logger.info(Thread.currentThread().getName() + ":doSomething start。。。");
                            System.out.println(Thread.currentThread().getName() + ":doSomething start。。。");
                            List<ExportOrder2Bo> list= queryOrderByBatch(sql);
                            //串行处理（重点二）
                            semaphore.acquire();
                            orderLists.add(list);
                            semaphore.release();
                            logger.info(Thread.currentThread().getName() + ":doSomething end-" + (System.currentTimeMillis()-tt2)+"|res list:"+list.size()+"|orderLists:"+orderLists.size());
                            System.out.println(Thread.currentThread().getName() + ":doSomething end-" + (System.currentTimeMillis()-tt2)+"|res list:"+list.size()+"|orderLists:"+orderLists.size());
                        } catch (Exception e) {
    //                        System.out.println("第"+sheetName+"批|SheetName:"+writeSheet.getSheetName()+"|SheetNo: "+writeSheet.getSheetNo());
                            System.out.println("-exportOrderInfo error Runnable：startIndex:"+startIndex+"|"+limitSizes+"|limitSizes|"+sql.toString());
                            logger.error("--------exportOrderInfo error--------:");
                            e.printStackTrace();
                        }
                    }
                }));
                sheetIndex++;
    //            System.out.println("开启sheetIndex："+sheetIndex+"|startIndex:"+startIndex+"|sql|"+sql.toString());
            }
    
                //阻塞等待返回
                try{
                    for(int i=0;i<headerflist.size();i++){
                        headerflist.get(i).get();
                        System.out.println("阻塞等待返回|"+i);
                    }
                    int len=orderLists.size();
                    logger.info("--------阻塞等待结束----------|分批实为："+sheetIndex+"|将执行写入excel size|"+len);
                    System.out.println("--------阻塞等待结束----------|分批实为："+sheetIndex+"|将执行写入excel size|"+len);
                    //并发写
                    /**
                     *2: 创建sheet
                     */
                    // 方法2 如果写到不同的sheet 同一个对象
                    String time = DateUtil.getSystemDate(DateUtil.DATE_TYPE1);
                    String fileName = URLEncoder.encode("exportOrder" + time, "UTF-8");
                    response.setHeader("Content-disposition", "attachment;filename=" + fileName + ".xlsx");
                    ExcelWriter excelWriter = EasyExcel.write(response.getOutputStream(), ExportOrder2Bo.class).build();
                    for (int i = 0; i < len; i++) {
                        long s3=System.currentTimeMillis();
                        String sheetName="sheet"+i;
                        WriteSheet writeSheet = EasyExcel.writerSheet(i, sheetName).head(ExportOrder2Bo.class).build();
                        System.out.println();
                        excelWriter.write(orderLists.get(i), writeSheet);
                        long s4=System.currentTimeMillis();
                        logger.info("第"+i+"批|SheetName:"+writeSheet.getSheetName()+"|SheetNo: "+writeSheet.getSheetNo()+"|excelWriter|"+(s4-s3));
                        System.out.println("第"+i+"批|SheetName:"+writeSheet.getSheetName()+"|SheetNo: "+writeSheet.getSheetNo()+"|excelWriter|"+(s4-s3));
                    }
                    /// 千万别忘记finish 会帮忙关闭流
                    excelWriter.finish();
                    System.out.println("--帮忙关闭流----");
                }catch(Exception e){
                    response.setStatus(400);
                    logger.error("--------process callback error--------:", e);
                }
                long s2=System.currentTimeMillis();
    
                // 这里注意 有同学反应使用swagger 会导致各种问题，请直接用浏览器或者用postman
                response.setContentType("application/vnd.ms-excel");
                response.setCharacterEncoding("utf-8");
                // 这里URLEncoder.encode可以防止中文乱码 当然和easyexcel没有关系
                System.out.println("并行处理订单耗时|"+(s2-s1));
            logger.info("并行处理订单耗时|"+(s2-s1));
            }catch (Exception e){
                logger.info("exportOrderInfo error!");
                e.printStackTrace();
                response.setStatus(400);
            }
            long s2=System.currentTimeMillis();
            System.out.println("导出订单耗时|"+(s2-s1));
            logger.info("导出订单耗时|"+(s2-s1));
            return vo;
        }
        /**
         * 根据条件查询订单
         * @param sql sql语句
         */
    private List<ExportOrder2Bo> queryOrderByBatch(Query sql){
    //
    //    Aggregation agg = Aggregation.newAggregation(
    //            Aggregation.match(Criteria.where("operatorId").is(dto.getOperatorId()).and("EndTime").gte(yesterdayStart).lte(yesterdayEnd)),
    //            Aggregation.s
    //            Aggregation.group("connectorSeq", "equipmentSeq", "stationSeq", "areaCode", "operatorId", "orgId").count().as("chargeCount")
    //                    .sum("TotalPower").as("chargePower").sum("costTime").as("chargeTime"));
    //    AggregationResults<StationElectricityDO> aggResults = mongoTemplate.aggregate(agg, "notificationChargeOrderInfo", StationElectricityDO.class);
    //    Query query=new Query();
    //    /**
    //     * 2: 拼写mongo 查询sql及 格式化list data 返回vo
    //     */
    //    List<StationElectricityDO> reslist = aggResults.getMappedResults();
    //
        List<ExportOrder2Bo> list = mongoTemplate.find(query,ExportOrder2Bo.class,"notificationChargeOrderInfo").skip(startIndex).limit(limitSizes);
        getCollection().find(sql)
    //    DBObject project = new BasicDBObject("$project", new BasicDBObject("DataTime", 1).append("ConnectorID", 1).append("orgName", 1).append("stationName", 1));
    
        long s1=System.currentTimeMillis();
    //    DBObject projection = new BasicDBObject(new BasicDBObject("StartChargeSeq", 1)
    //            .append("ConnectorID", 1)
    //            .append("CreatedTime", 1)
    //            .append("StartTime", 1)
    //            .append("EndTime", 1)
    //            .append("costTime", 1)
    //            .append("TotalPower", 1)
    //            .append("TotalElecMoney", 1)
    //            .append("TotalSeviceMoney", 1)
    //            .append("TotalMoney", 1)
    //            .append("equipmentId", 1)
    //            .append("areaName", 1)
    //            .append("orgName", 1)
    //            .append("stationName", 1));
        System.out.println(Thread.currentThread().getName() + "|mongo sql"+JSON.toJSON(sql));
    //    DBCursor output = mongoTemplate.getCollection("notificationChargeOrderInfo").find(sql,projection).skip(startIndex).limit(limitSizes);
        List<ExportOrder2Bo> list=mongoTemplate.find(sql,ExportOrder2Bo.class,"notificationChargeOrderInfo");//5s +_6s=11  6s
    //    DBObject next = null;
        long s2=System.currentTimeMillis();
    //    List<ExportOrder2Bo> list = new ArrayList<ExportOrder2Bo>();
    //    while (output.hasNext()) {
    //        next = output.next();
    //        ExportOrder2Bo obj = new ExportOrder2Bo();
    //        String StartChargeSeq = (String) next.get("StartChargeSeq");
    //        Date startTime = (Date) next.get("StartTime");
    //        Date endTime = (Date) next.get("EndTime");
    //        String equipmentId = (String) next.get("equipmentId");
    //        String ConnectorID = (String) next.get("ConnectorID");
    //        String stationName = (String) next.get("stationName");
    //        String orgName = (String) next.get("orgName");
    //        Double TotalPower = (Double) next.get("TotalPower");
    //        Double TotalElecMoney = (Double) next.get("TotalElecMoney");
    //        Double TotalSeviceMoney = (Double) next.get("TotalSeviceMoney");
    //        Double TotalMoney = (Double) next.get("TotalMoney");
    //        if(next.get("costTime")!=null){
    //            int costTime = (int) next.get("costTime");
    //            double times = new BigDecimal(costTime).divide(BigDecimal.valueOf(3600),2, BigDecimal.ROUND_HALF_UP).doubleValue();
    //            obj.setChargeTime(times);
    //        }
    //        obj.setOrderId(StartChargeSeq);
    //        obj.setDateTime((Date) next.get("CreatedTime"));
    //        obj.setStubId(equipmentId);
    //        obj.setStationName(stationName);
    //        obj.setOrgName(orgName);
    //        obj.setTimeStart(startTime);
    //        obj.setTimeEnd(endTime);
    //        obj.setConnectorID(ConnectorID);
    //        obj.setCity((String) next.get("areaName"));
    //        obj.setElectric(TotalPower);
    //        obj.setFeeElectric(TotalElecMoney);
    //        obj.setFeeService(TotalSeviceMoney);
    //        obj.setFeeTotal(TotalMoney);
    //        list.add(obj);
    //    }
    //    if (list.size() == 0) {
    //        logger.info("|导出excel查询数据库为空");
    //        System.out.println("|导出excel查询数据库为空");
    //        return list;
    //    }
    //    if (excelWriter == null) {
    //        System.err.println("excelWriter is null");
    //    }
    //    long s3=System.currentTimeMillis();
        logger.info(Thread.currentThread().getName() + "|查询结果list|"+list.size()+"|mongo sql time|"+(s2-s1));
        System.out.println(Thread.currentThread().getName() + "|查询结果list|"+list.size()+"|mongo sql time|"+(s2-s1));
    //    logger.info("查询结果list|"+list.size());
        WriteSheet writeSheetSync=(WriteSheet)sheetMap.get(writeSheet.getSheetNo());
        excelWriter.write(list,writeSheetSync);
    //    // 分页去数据库查询数据 这里可以去数据库查询每一页的数据
    //    long s4=System.currentTimeMillis();
    //    System.out.println("startIndex"+startIndex+"|mongo sql time|"+(s2-s1)+"|hasNext|"+(s3-s2)+"|excelWriter|"+(s4-s3));
    //    logger.info("startIndex"+startIndex+"|mongo sql time|"+(s2-s1)+"|hasNext|"+(s3-s2)+"|excelWriter|"+(s4-s3));
        return list;
    //    String excelTempletPath = RedisKeysDefine.EXPORT_ORDER_EXCEL;
    }

[java_Semaphore]: https://www.cnblogs.com/klbc/p/9500947.html