Merge pull request '抽数：验证mysql和mongo数量是否一致' (#68) from xinwei into master

Reviewed-on: #68

yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/dal/mongodb/checkticket/CheckTicketTodayRepository.java

@@ -12,24 +12,29 @@ import java.util.List;
 
 
 @Repository
-public interface CheckTicketTodayRepository extends MongoRepository<CheckTicketToday,String> {
+public interface CheckTicketTodayRepository extends MongoRepository<CheckTicketToday, String> {
     /**
      * 根据dataId数组查找
+     *
      * @param dataIdList
      * @return java.util.List<cn.iocoder.yudao.module.infra.dal.dataobject.checkticket.CheckTicket>
      */
     @Query(value = "{ 'dataId': { $in: ?0 } }", fields = "{ 'dataId': 1, '_id': 0 }")
     List<CheckTicketToday> findByDataIdIn(List<String> dataIdList);
+
     /**
      * 找到所有不在时间段内的日期字段
+     *
      * @param startDate
      * @param endDate
      * @return java.util.List<cn.iocoder.yudao.module.infra.dal.dataobject.checkticket.CheckTicketToday>
      */
     @Query("{ 'checkticketdate': { $not: { $gte: ?0, $lt: ?1 } } }")
     List<CheckTicketToday> findNotInTimeRange(String startDate, String endDate);
+
     /**
      * 查询所有去重的 checkticketdate 字段值
+     *
      * @return 去重后的 checkticketdate 字段值列表
      */
     @Aggregation(pipeline = {
@@ -39,9 +44,10 @@ public interface CheckTicketTodayRepository extends MongoRepository<CheckTicketT
     List<String> findDistinctDate();
 
     /**
-     * 删除 checkticketdate 字段值在给定列表中的所有记录
-     * @param dates 日期列表
+     * 计算自定义日期的总数
+     * @param date
+     * @return int
      */
-    @Query("{ 'checkticketdate': { $in: ?0 } }")
-    void deleteByCheckticketdateIn(List<String> dates);
+    @Query(value = "{ 'checkticketdate': ?0 }", count = true)
+    long  countByCustomDate(String date);
 }

yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/dal/mongodb/saledata/SaleDataTodayRepository.java

@@ -87,4 +87,12 @@ public interface SaleDataTodayRepository extends MongoRepository<SaleDataToday,S
             "{$project: {_id: 0, sddate: '$_id'}}"
     })
     List<String> findDistinctDate();
+
+    /**
+     * 根据日期计算总数
+     * @param date
+     * @return java.lang.Long
+     */
+    @Query(value = "{ 'sddate': ?0 }", count = true)
+    Long countByCustomDate(String date);
 }

yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/job/ticketing/TicketDataMigration.java

@@ -63,7 +63,11 @@ public class TicketDataMigration {
         int failedCount = 0;
         int pageNumber = 1;
         XxlJobHelper.log("抽数服务：mysql销售数据开始抽数...");
-        TicketingSamplingRespVO ticketingSamplingRespVO = this.getUrlResponseData(SALE_DATA_SERVICE, SALE_DATA_METHOD, TickingDateUtils.getNowDayFormat(), pageNumber);
+        TicketingSamplingRespVO ticketingSamplingRespVO = this.getUrlResponseData(SALE_DATA_SERVICE, SALE_DATA_METHOD, TickingDateUtils.getPreviousDayFormat(), pageNumber);
+        if (ticketingSamplingRespVO == null) {
+            XxlJobHelper.log("暂无数据！");
+            return;
+        }
         int totalRows = ticketingSamplingRespVO.getTotalRows();
         List<Map<String, Object>> mapList = ticketingSamplingRespVO.getDataMapList();
         // 总条数小于等于每页条数 则直接按照总条数进行插入即可
@@ -78,7 +82,7 @@ public class TicketDataMigration {
             int totalPages = ticketingSamplingRespVO.getTotalPages();
             // 处理剩余页的数据
             for (; pageNumber <= totalPages; pageNumber++) {
-                ticketingSamplingRespVO = this.getUrlResponseData(SALE_DATA_SERVICE, SALE_DATA_METHOD, TickingDateUtils.getNowDayFormat(), pageNumber);
+                ticketingSamplingRespVO = this.getUrlResponseData(SALE_DATA_SERVICE, SALE_DATA_METHOD, TickingDateUtils.getPreviousDayFormat(), pageNumber);
 
                 mapList = ticketingSamplingRespVO.getDataMapList();
                 // 对最后一页进行处理，得到最后一页的实际条数
@@ -90,7 +94,12 @@ public class TicketDataMigration {
             }
 
         }
-
+        if (!saleDataService.dataValidation(TickingDateUtils.getPreviousDayFormat())) {
+            XxlJobHelper.log("mysql和mongo数据验证不一致！！！");
+            XxlJobHelper.log("重新抽取({})mongo数据", TickingDateUtils.getPreviousDayFormat());
+            getSaleDataByMongoDBHandler();
+            XxlJobHelper.log("抽数完，比对结果：{}", saleDataService.dataValidation(TickingDateUtils.getPreviousDayFormat()) ? "相等" : "不想等");
+        }
         XxlJobHelper.log("抽数服务：销售数据抽数正常结束！总共：{}条数据，{}条数据插入失败！", totalRows, failedCount);
     }
 
@@ -102,7 +111,11 @@ public class TicketDataMigration {
         int failedCount = 0;
         int pageNumber = 1;
         XxlJobHelper.log("抽数服务：mysql检票数据开始抽数...");
-        TicketingSamplingRespVO ticketingSamplingRespVO = this.getUrlResponseData(CHECK_TICKET_SERVICE, CHECK_TICKET_METHOD, TickingDateUtils.getNowDayFormat(), pageNumber);
+        TicketingSamplingRespVO ticketingSamplingRespVO = this.getUrlResponseData(CHECK_TICKET_SERVICE, CHECK_TICKET_METHOD, TickingDateUtils.getPreviousDayFormat(), pageNumber);
+        if (ticketingSamplingRespVO == null) {
+            XxlJobHelper.log("暂无数据！");
+            return;
+        }
         int totalRows = ticketingSamplingRespVO.getTotalRows();
         List<Map<String, Object>> mapList = ticketingSamplingRespVO.getDataMapList();
         // 总条数小于等于每页条数 则直接按照总条数进行插入即可
@@ -117,7 +130,7 @@ public class TicketDataMigration {
             int totalPages = ticketingSamplingRespVO.getTotalPages();
             // 处理剩余页的数据
             for (; pageNumber <= totalPages; pageNumber++) {
-                ticketingSamplingRespVO = this.getUrlResponseData(CHECK_TICKET_SERVICE, CHECK_TICKET_METHOD, TickingDateUtils.getNowDayFormat(), pageNumber);
+                ticketingSamplingRespVO = this.getUrlResponseData(CHECK_TICKET_SERVICE, CHECK_TICKET_METHOD, TickingDateUtils.getPreviousDayFormat(), pageNumber);
                 mapList = ticketingSamplingRespVO.getDataMapList();
                 // 对最后一页进行处理，得到最后一页的实际条数
                 if (pageNumber != totalPages) {
@@ -126,9 +139,13 @@ public class TicketDataMigration {
                     failedCount += insertCheckTicket(mapList, totalRows % PAGE_SIZE, pageNumber);
                 }
             }
-
         }
-
+        if (!checkTicketService.dataValidation(TickingDateUtils.getPreviousDayFormat())) {
+            XxlJobHelper.log("mysql和mongo数据验证不一致！！！");
+            XxlJobHelper.log("重新抽取({})mongo数据", TickingDateUtils.getPreviousDayFormat());
+            getCheckTicketByMongoDBHandler();
+            XxlJobHelper.log("抽数完，比对结果：{}", checkTicketService.dataValidation(TickingDateUtils.getPreviousDayFormat()) ? "相等" : "不想等");
+        }
         XxlJobHelper.log("抽数服务：检票数据抽数正常结束！总共：{}条数据，{}条数据插入失败！", totalRows, failedCount);
     }
 
@@ -142,6 +159,10 @@ public class TicketDataMigration {
         int pageNumber = 1;
         XxlJobHelper.log("抽数服务：mongodb销售数据开始抽数...");
         TicketingSamplingRespVO ticketingSamplingRespVO = this.getUrlResponseData(SALE_DATA_SERVICE, SALE_DATA_METHOD, TickingDateUtils.getNowDayFormat(), pageNumber);
+        if (ticketingSamplingRespVO == null) {
+            XxlJobHelper.log("暂无数据！");
+            return;
+        }
         int totalRows = ticketingSamplingRespVO.getTotalRows();
         List<Map<String, Object>> mapList;
         // 获得总页数
@@ -173,6 +194,10 @@ public class TicketDataMigration {
         int pageNumber = 1;
         String date = TickingDateUtils.getNowDayFormat();
         TicketingSamplingRespVO ticketingSamplingRespVO = this.getUrlResponseData(CHECK_TICKET_SERVICE, CHECK_TICKET_METHOD, date, pageNumber);
+        if (ticketingSamplingRespVO == null) {
+            XxlJobHelper.log("暂无数据！");
+            return;
+        }
         int totalRows = ticketingSamplingRespVO.getTotalRows();
         List<Map<String, Object>> mapList;
         // 获得总页数
@@ -307,7 +332,7 @@ public class TicketDataMigration {
      * 删除（今年和去年）往前推31天外的数据和抽取去年明天的数据
      */
     @XxlJob("deleteTwoYearOneMonthAgoAndUpdateLastYearTomorrow")
-    public void deleteTwoYearOneMonthAgoAndUpdateLastYearTomorrow(){
+    public void deleteTwoYearOneMonthAgoAndUpdateLastYearTomorrow() {
         XxlJobHelper.log("删除多余数据开始...");
         // 检票数据今年去年多余删除
         XxlJobHelper.log("删除检票日期为{}", checkTicketService.deleteTwoYearLastMonthAgo().toString());
@@ -401,7 +426,7 @@ public class TicketDataMigration {
             int currentDay = startDate.get(Calendar.DAY_OF_MONTH);
             result.add(String.format("抽数服务：%d年%d月%d日：" + methodName + "数据抽数正常结束！总共：%d条数据，%d条数据重复，%d条数据插入成功！", currentYear, currentMonth, currentDay, totalRows, duplicatesCount, successCount));
             XxlJobHelper.log("抽数服务：{}年{}月{}日：{}数据抽数正常结束！总共：{}条数据，{}条数据重复，{}条数据插入成功！", currentYear, currentMonth, currentDay, methodName, totalRows, duplicatesCount, successCount);
-            if (startDate.equals(endDate)){ // 如果日期相等，退出循环
+            if (startDate.equals(endDate)) { // 如果日期相等，退出循环
                 break;
             }
             // 日期递增一天
@@ -488,25 +513,20 @@ public class TicketDataMigration {
             XxlJobHelper.handleFail("数据抽数失败,请重试");
             throw new RuntimeException();
         } else if (ticketingSamplingRespVO.getPageNumber() == null || ticketingSamplingRespVO.getPageSize() == null || ticketingSamplingRespVO.getTotalPages() == null || ticketingSamplingRespVO.getTotalRows() == null || ticketingSamplingRespVO.getDataMapList().isEmpty()) {
-            XxlJobHelper.log("{}，请求抽数成功，但返回参数值存在问题：{}", date, ticketingSamplingRespVO);
+            XxlJobHelper.log("{}，请求抽数成功，但返回参数无值：{}", date, ticketingSamplingRespVO);
             return null;
         }
         return ticketingSamplingRespVO;
     }
 
     /**
-     * 请求失败消息
+     * 校验数据
      *
-     * @param
-     * @return void
+     * @param date
+     * @return java.lang.Boolean
      */
-    private void requestFailedMessage(TicketingSamplingRespVO ticketingSamplingRespVO) {
-        // 如果访问失败则让调度中心重新调用
-        if (!ticketingSamplingRespVO.isPassflag()) {
-            // 让调度中心重新调用
-            XxlJobHelper.log("抽数服务：检票数据抽数失败，返回消息：" + ticketingSamplingRespVO.getRtnMsg());
-            XxlJobHelper.handleFail("销售数据抽数失败,请重试");
-        }
+    public Boolean dataValidation(String date) {
+        return checkTicketService.dataValidation(date) && saleDataService.dataValidation(date);
     }
 
 }

yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/service/checkticket/CheckTicketService.java

@@ -56,4 +56,11 @@ public interface CheckTicketService {
     List<String> deleteTwoYearLastMonthAgo();
 
     List<String> findDistinctDate();
+
+    /**
+     * mongo和mysql数据数量验证
+     * @param date
+     * @return java.lang.Boolean
+     */
+    Boolean dataValidation(String date);
 }

yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/service/checkticket/CheckTicketServiceImpl.java

@@ -149,4 +149,9 @@ public class CheckTicketServiceImpl implements CheckTicketService {
         return checkTicketTodayRepository.findDistinctDate();
     }
 
+    @Override
+    public Boolean dataValidation(String date) {
+        return checkTicketMapper.selectCount(CheckTicketDO::getCheckticketdate, date) == checkTicketTodayRepository.countByCustomDate(date);
+    }
+
 }

yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/service/saledata/SaleDataService.java

@@ -55,4 +55,11 @@ public interface SaleDataService {
      * @return java.util.List<java.lang.String>
      */
     List<String> deleteTwoYearLastMonthAgo();
+
+    /**
+     * mongo和mysql数据数量验证
+     * @param date
+     * @return boolean
+     */
+    Boolean dataValidation(String date);
 }

yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/service/saledata/SaleDataServiceImpl.java

@@ -136,4 +136,9 @@ public class SaleDataServiceImpl implements SaleDataService {
         mongoTemplate.remove(query, "sale_data_today");
         return filteredDate;
     }
+
+    @Override
+    public Boolean dataValidation(String date) {
+        return Objects.equals(saleDataMapper.selectCount(SaleDataDO::getSddate, date), saleDataTodayRepository.countByCustomDate(date));
+    }
 }