|
@@ -1,16 +1,20 @@
|
|
|
package com.sckw.report.consumer;
|
|
package com.sckw.report.consumer;
|
|
|
|
|
|
|
|
-import com.alibaba.fastjson.JSON;
|
|
|
|
|
-import com.alibaba.fastjson.JSONObject;
|
|
|
|
|
|
|
+import com.alibaba.fastjson2.JSON;
|
|
|
|
|
+import com.alibaba.fastjson2.JSONObject;
|
|
|
|
|
+import com.sckw.core.utils.StringUtils;
|
|
|
|
|
+import com.sckw.mongo.model.SckwLogisticsOrder;
|
|
|
import com.sckw.mongo.model.SckwTradeOrder;
|
|
import com.sckw.mongo.model.SckwTradeOrder;
|
|
|
|
|
+import com.sckw.mongo.model.SckwWaybillOrder;
|
|
|
|
|
+import com.sckw.report.dao.SckwLogisticsOrderRepository;
|
|
|
import com.sckw.report.dao.SckwTradeOrderRepository;
|
|
import com.sckw.report.dao.SckwTradeOrderRepository;
|
|
|
|
|
+import com.sckw.report.dao.SckwWaybillOrderRepository;
|
|
|
|
|
+import com.sckw.stream.model.SckwBusSum;
|
|
|
import lombok.RequiredArgsConstructor;
|
|
import lombok.RequiredArgsConstructor;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
|
import org.springframework.context.annotation.Bean;
|
|
import org.springframework.context.annotation.Bean;
|
|
|
import org.springframework.context.annotation.Configuration;
|
|
import org.springframework.context.annotation.Configuration;
|
|
|
-import java.util.Date;
|
|
|
|
|
-import java.util.Map;
|
|
|
|
|
import java.util.function.Consumer;
|
|
import java.util.function.Consumer;
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
@@ -24,12 +28,20 @@ import java.util.function.Consumer;
|
|
|
public class SckwBusSumConsumer {
|
|
public class SckwBusSumConsumer {
|
|
|
@Autowired
|
|
@Autowired
|
|
|
SckwTradeOrderRepository tradeOrderRepository;
|
|
SckwTradeOrderRepository tradeOrderRepository;
|
|
|
|
|
+ @Autowired
|
|
|
|
|
+ SckwLogisticsOrderRepository logisticsOrderRepository;
|
|
|
|
|
+ @Autowired
|
|
|
|
|
+ SckwWaybillOrderRepository waybillOrderRepository;
|
|
|
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * @desc 消息订阅
|
|
|
|
|
+ * @author zk
|
|
|
|
|
+ * @date 2023/7/8
|
|
|
|
|
+ **/
|
|
|
@Bean
|
|
@Bean
|
|
|
- public Consumer<Map> sckwBusSum() {
|
|
|
|
|
|
|
+ public Consumer<SckwBusSum> sckwBusSum() {
|
|
|
return sckwBusSum -> {
|
|
return sckwBusSum -> {
|
|
|
try {
|
|
try {
|
|
|
- System.out.println(new Date()+"---------"+JSON.toJSONString(sckwBusSum));
|
|
|
|
|
addMongodb(sckwBusSum);
|
|
addMongodb(sckwBusSum);
|
|
|
} catch (Exception e) {
|
|
} catch (Exception e) {
|
|
|
log.error("处理业务数据汇总消息:{}异常", JSON.toJSONString(sckwBusSum), e);
|
|
log.error("处理业务数据汇总消息:{}异常", JSON.toJSONString(sckwBusSum), e);
|
|
@@ -37,8 +49,45 @@ public class SckwBusSumConsumer {
|
|
|
};
|
|
};
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- public void addMongodb(Map<String, Object> sckwBusSum) {
|
|
|
|
|
- SckwTradeOrder order = JSONObject.parseObject(JSON.toJSONString(sckwBusSum), SckwTradeOrder.class);
|
|
|
|
|
- tradeOrderRepository.insert(order);
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * @param busSum rabbitMq业务汇总数据发送/消费对象
|
|
|
|
|
+ * @desc 业务存储处理
|
|
|
|
|
+ * @author zk
|
|
|
|
|
+ * @date 2023/7/8
|
|
|
|
|
+ **/
|
|
|
|
|
+ public void addMongodb(SckwBusSum busSum) {
|
|
|
|
|
+ /**初级校验**/
|
|
|
|
|
+ if (busSum.getObject() == null || StringUtils.isBlank(busSum.getBusSumType()) || busSum.getMethod() == 0) {
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**业务存储**/
|
|
|
|
|
+ if (busSum.getBusSumType().equals("tradeOrder")) {
|
|
|
|
|
+ SckwTradeOrder order = JSONObject.parseObject(JSON.toJSONString(busSum.getObject()), SckwTradeOrder.class);
|
|
|
|
|
+ if (busSum.getMethod() == 1) {
|
|
|
|
|
+ tradeOrderRepository.insert(order);
|
|
|
|
|
+ } else {
|
|
|
|
|
+ tradeOrderRepository.save(order);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (busSum.getBusSumType().equals("logisticsOrder")) {
|
|
|
|
|
+ SckwLogisticsOrder order = JSONObject.parseObject(JSON.toJSONString(busSum.getObject()), SckwLogisticsOrder.class);
|
|
|
|
|
+ if (busSum.getMethod() == 1) {
|
|
|
|
|
+ logisticsOrderRepository.insert(order);
|
|
|
|
|
+ } else {
|
|
|
|
|
+ logisticsOrderRepository.save(order);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (busSum.getBusSumType().equals("waybillOrder")) {
|
|
|
|
|
+ SckwWaybillOrder order = JSONObject.parseObject(JSON.toJSONString(busSum.getObject()), SckwWaybillOrder.class);
|
|
|
|
|
+ if (busSum.getMethod() == 1) {
|
|
|
|
|
+ waybillOrderRepository.insert(order);
|
|
|
|
|
+ } else {
|
|
|
|
|
+ waybillOrderRepository.save(order);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
}
|
|
}
|