消费巡查消息调整

epmet-module/open-data-worker/open-data-worker-server/src/main/java/com/epmet/opendata/mq/listener/OpenDataPatrolChangeEventListener.java

@@ -6,6 +6,8 @@ import com.epmet.commons.rocketmq.messages.StaffPatrolMQMsg;
 import com.epmet.commons.tools.constant.NumConstant;
 import com.epmet.commons.tools.constant.StrConstant;
 import com.epmet.commons.tools.distributedlock.DistributedLock;
+import com.epmet.commons.tools.exception.EpmetErrorCode;
+import com.epmet.commons.tools.exception.EpmetException;
 import com.epmet.commons.tools.exception.ExceptionUtils;
 import com.epmet.commons.tools.exception.RenException;
 import com.epmet.commons.tools.redis.RedisKeys;
@@ -77,10 +79,6 @@ public class OpenDataPatrolChangeEventListener implements MessageListenerConcurr
         if (!StrConstant.PY_CUSTOMER.equals(msgObj.getCustomerId())) {
             return;
         }
-        if (msgObj == null) {
-            log.warn("consumeMessage msg body is blank");
-            return;
-        }
         RLock lock = null;
         try {
             lock = distributedLock.getLock(String.format("lock:open_data_patrol:%s:%s",tags, msgObj.getObjectId()),
@@ -104,6 +102,9 @@ public class OpenDataPatrolChangeEventListener implements MessageListenerConcurr
                     log.error("错误的消息类型:{}", tags);
 
             }
+            if (aBoolean == null || !aBoolean){
+                throw new EpmetException(EpmetErrorCode.EPMET_COMMON_OPERATION_FAIL.getCode(),"消费失败!");
+            }
 
         } catch (RenException e) {
             // 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试

epmet-module/open-data-worker/open-data-worker-server/src/main/java/com/epmet/opendata/service/impl/UserPatrolRecordServiceImpl.java

@@ -71,12 +71,8 @@ public class UserPatrolRecordServiceImpl extends BaseServiceImpl<UserPatrolRecor
         }
         List<MidPatrolRecordResult> data = record.getData();
         if (CollectionUtils.isEmpty(data)) {
-            //数据已被删除了
+            //数据还未被插入
-            //暂时设置error 用于排错
+            throw new EpmetException(EpmetErrorCode.SERVER_ERROR.getCode(),"获取巡查记录返回为空，param:"+ JSON.toJSONString(midPatrolFormDTO));
-            log.error("insertPatrolRecord获取巡查记录返回为空，param:{}", JSON.toJSONString(midPatrolFormDTO));
-            int effectRow = baseDao.deleteById(patrolRecordForm.getId());
-            log.warn("del effectRow:{}", effectRow);
-            throw new EpmetException(EpmetErrorCode.SERVER_ERROR.getCode());
         }
         List<UserPatrolRecordEntity> insertList = new ArrayList<>();
         data.forEach(o-> insertList.add(buildEntity(o)));
@@ -103,10 +99,7 @@ public class UserPatrolRecordServiceImpl extends BaseServiceImpl<UserPatrolRecor
         if (CollectionUtils.isEmpty(data)) {
             //数据已被删除了
             //暂时设置error 用于排错
-            log.error("updatePatrolRecord 获取巡查记录返回为空，param:{}", JSON.toJSONString(midPatrolFormDTO));
+            throw new EpmetException(EpmetErrorCode.SERVER_ERROR.getCode(),"获取巡查记录返回为空，param:"+ JSON.toJSONString(midPatrolFormDTO));
-            int effectRow = baseDao.deleteById(patrolRecordForm.getId());
-            log.warn("del effectRow:{}", effectRow);
-            throw new EpmetException(EpmetErrorCode.SERVER_ERROR.getCode());
         }
         data.forEach(o->{
             UserPatrolRecordEntity recordEntity = buildEntity(o);