Pu Zhibing
2025-04-30 1adec9fead03f0f788a73f9349ccba86569e31f3
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
package com.ruoyi.integration.rocket.listener;
 
import com.ruoyi.chargingPile.api.feignClient.AccountingStrategyDetailClient;
import com.ruoyi.chargingPile.api.feignClient.ChargingGunClient;
import com.ruoyi.chargingPile.api.feignClient.FaultMessageClient;
import com.ruoyi.chargingPile.api.model.TAccountingStrategyDetail;
import com.ruoyi.chargingPile.api.model.TChargingGun;
import com.ruoyi.chargingPile.api.model.TFaultMessage;
import com.ruoyi.chargingPile.api.vo.GetChargingGunByCode;
import com.ruoyi.integration.api.feignClient.TCECClient;
import com.ruoyi.integration.api.model.Online;
import com.ruoyi.integration.api.model.UploadRealTimeMonitoringData;
import com.ruoyi.integration.iotda.constant.SendTagConstant;
import com.ruoyi.integration.mongodb.service.UploadRealTimeMonitoringDataService;
import com.ruoyi.integration.rocket.model.ChargingMessage;
import com.ruoyi.integration.rocket.model.ChargingOrderMessage;
import com.ruoyi.integration.rocket.model.UploadRealTimeMonitoringDataMessage;
import com.ruoyi.integration.rocket.produce.EnhanceProduce;
import com.ruoyi.integration.rocket.util.EnhanceMessageHandler;
import com.ruoyi.order.api.feignClient.ChargingOrderClient;
import com.ruoyi.order.api.model.TChargingOrder;
import com.ruoyi.order.api.query.UploadRealTimeMonitoringDataQuery;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
 
import javax.annotation.Resource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.Objects;
 
@Slf4j
@Component
@RocketMQMessageListener(
        messageModel = MessageModel.CLUSTERING,
        consumerGroup = "charge_upload_real_time_monitoring_data",
        topic = "charge_upload_real_time_monitoring_data",
        selectorExpression = "upload_real_time_monitoring_data"
)
public class UploadRealTimeMonitoringDataMessageListener implements RocketMQListener<UploadRealTimeMonitoringDataMessage> {
 
    @Autowired
    private UploadRealTimeMonitoringDataService uploadRealTimeMonitoringDataService;
    
    @Resource
    private ChargingOrderClient chargingOrderClient;
    @Resource
    private AccountingStrategyDetailClient accountingStrategyDetailClient;
    @Autowired
    private EnhanceProduce enhanceProduce;
 
 
    
    
    
 
    /**
     * 监听消费消息,不需要执行业务处理,委派给父类做基础操作,父类做完基础操作后会调用子类的实际处理类型
     */
    @Override
    public void onMessage(UploadRealTimeMonitoringDataMessage message) {
        try {
            log.info("上传实时监测数据-业务消息处理:{}",message);
            // 持久化消息
            UploadRealTimeMonitoringData uploadRealTimeMonitoringData = new UploadRealTimeMonitoringData();
            BeanUtils.copyProperties(message,uploadRealTimeMonitoringData);
            // 查询mogondb上一条数据
            UploadRealTimeMonitoringData data = uploadRealTimeMonitoringDataService.getLastDataById(message.getTransaction_serial_number());
            // 查询订单
            TChargingOrder chargingOrder = chargingOrderClient.getOrderByCode(message.getTransaction_serial_number()).getData();
            // 查询当前时间段的计费策略
            TAccountingStrategyDetail accountingStrategyDetail = accountingStrategyDetailClient.getDetailBySiteId(chargingOrder.getSiteId()).getData();
            uploadRealTimeMonitoringData.setElectrovalence_all(accountingStrategyDetail.getElectrovalence());
            uploadRealTimeMonitoringData.setService_charge(accountingStrategyDetail.getServiceCharge());
            if (Objects.nonNull(data)) {
                uploadRealTimeMonitoringDataService.updateById(data.getId());
                uploadRealTimeMonitoringData.setPeriod_electric_price(message.getPaid_amount().subtract(data.getPaid_amount()));
                uploadRealTimeMonitoringData.setPeriod_charging_degree(message.getCharging_degree().subtract(data.getCharging_degree()));
                uploadRealTimeMonitoringData.setPeriod_service_price(message.getCharging_degree().multiply(accountingStrategyDetail.getServiceCharge()).setScale(4, RoundingMode.HALF_UP));
            }else {
                log.info("首次上传实时监测数据");
                uploadRealTimeMonitoringData.setPeriod_electric_price(message.getPaid_amount());
                uploadRealTimeMonitoringData.setPeriod_charging_degree(message.getCharging_degree());
                uploadRealTimeMonitoringData.setPeriod_service_price(message.getCharging_degree().multiply(accountingStrategyDetail.getServiceCharge()).setScale(4, RoundingMode.HALF_UP));
            }
            uploadRealTimeMonitoringData.setOrderType(chargingOrder.getOrderType());
            uploadRealTimeMonitoringData.setSiteId(chargingOrder.getSiteId());
            uploadRealTimeMonitoringData.setStatus(chargingOrder.getStatus());
            int i = uploadRealTimeMonitoringDataService.create(uploadRealTimeMonitoringData);
            if(i == 0){
                log.error("数据存储mongo失败");
            }
        
            // 业务处理
            UploadRealTimeMonitoringDataQuery query = new UploadRealTimeMonitoringDataQuery();
            BeanUtils.copyProperties(uploadRealTimeMonitoringData, query);
            chargingOrderClient.chargeMonitoring(query);
            // 订单id
            ChargingOrderMessage chargingOrderMessage3 = new ChargingOrderMessage();
            chargingOrderMessage3.setOrderNumber(chargingOrder.getCode());
            // 推送充电订单信息
            ChargingMessage chargingMessage4 = new ChargingMessage();
            chargingMessage4.setServiceId(SendTagConstant.ORDER_STATUS);
            chargingMessage4.setOrderMessage(chargingOrderMessage3);
            enhanceProduce.orderInfoMessage(chargingMessage4);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}