mq对列修改

xuhy

2024-09-13 c2060ca066394dd10b8b2b7f9ec486f5df099a24

mq对列修改

 ruoyi-api/ruoyi-api-order/src/main/java/com/ruoyi/order/api/model/TChargingOrder.java

@@ -186,7 +186,9 @@
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    @TableField("refund_time")
    private LocalDateTime refundTime;

    @ApiModelProperty(value = "总电量")
    @TableField("total_electricity")
    private BigDecimal totalElectricity;// 总电量

    @ApiModelProperty(value = "支付时间")
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")

 ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/iotda/utils/listener/IotMessageListener.java

@@ -63,7 +63,7 @@
        String result;
        switch (service_id){
            case SendTagConstant.ONLINE:
                OnlineMessage onlineMessage = JSON.parseObject(jsonObject.toJSONString(),OnlineMessage.class);
                OnlineMessage onlineMessage = JSON.parseObject(content.toJSONString(),OnlineMessage.class);
                sendResult = enhanceProduce.onlineMessage(onlineMessage);
                // 响应硬件
                // 业务处理 登录认证应答
@@ -74,7 +74,7 @@
                log.info("充电桩登录认证-返回结果：{}",result);
                break;
            case SendTagConstant.PING:
                PingMessage pingMessage = JSON.parseObject(jsonObject.toJSONString(),PingMessage.class);
                PingMessage pingMessage = JSON.parseObject(content.toJSONString(),PingMessage.class);
                sendResult = enhanceProduce.pingMessage(pingMessage);
                // 响应硬件
                Pong pong = new Pong();
@@ -85,12 +85,12 @@
                log.info("充电桩心跳包-返回结果：{}",result);
                break;
            case SendTagConstant.END_CHARGE:
                EndChargeMessage endChargeMessage = JSON.parseObject(jsonObject.toJSONString(),EndChargeMessage.class);
                EndChargeMessage endChargeMessage = JSON.parseObject(content.toJSONString(),EndChargeMessage.class);
                sendResult = enhanceProduce.endChargeMessage(endChargeMessage);
                // 响应硬件
                break;
            case SendTagConstant.BILLING_MODE_VERIFY:
                BillingModeVerifyMessage billingModeVerifyMessage = JSON.parseObject(jsonObject.toJSONString(),BillingModeVerifyMessage.class);
                BillingModeVerifyMessage billingModeVerifyMessage = JSON.parseObject(content.toJSONString(),BillingModeVerifyMessage.class);
                sendResult = enhanceProduce.billingModeVerifyMessage(billingModeVerifyMessage);
                // 响应硬件
                BillingModeVerifyReply billingModeVerifyReply = new BillingModeVerifyReply();
@@ -117,7 +117,7 @@
                iotMessageProduce.sendMessage(deviceId, ServiceIdMenu.BILLING_MODE_VERIFY_REPLY.getKey(),messageUtil.billingModeVerifyReply(billingModeVerifyReply));
                break;
            case SendTagConstant.ACQUISITION_BILLING_MODE:
                AcquisitionBillingModeMessage acquisitionBillingModeMessage = JSON.parseObject(jsonObject.toJSONString(),AcquisitionBillingModeMessage.class);
                AcquisitionBillingModeMessage acquisitionBillingModeMessage = JSON.parseObject(content.toJSONString(),AcquisitionBillingModeMessage.class);
                sendResult = enhanceProduce.acquisitionBillingModeMessage(acquisitionBillingModeMessage);
                // 响应硬件   计费模型请求应答  1=尖阶段，2=峰阶段，3=平阶段，4=谷阶段
                List<TAccountingStrategyDetail> accountingStrategyDetails = accountingStrategyDetailClient.getDetailListByCode(acquisitionBillingModeMessage.getCharging_pile_code()).getData();
@@ -130,55 +130,55 @@
                iotMessageProduce.sendMessage(deviceId, ServiceIdMenu.ACQUISITION_BILLING_MODE_REPLY.getKey(),messageUtil.acquisitionBillingModeReply(acquisitionBillingModeReply));
                break;
            case SendTagConstant.UPLOAD_REAL_TIME_MONITORING_DATA:
                UploadRealTimeMonitoringDataMessage uploadRealTimeMonitoringDataMessage = JSON.parseObject(jsonObject.toJSONString(),UploadRealTimeMonitoringDataMessage.class);
                UploadRealTimeMonitoringDataMessage uploadRealTimeMonitoringDataMessage = JSON.parseObject(content.toJSONString(),UploadRealTimeMonitoringDataMessage.class);
                sendResult = enhanceProduce.uploadRealTimeMonitoringDataMessage(uploadRealTimeMonitoringDataMessage);
                // 响应硬件
                break;
            case SendTagConstant.CHARGING_HANDSHAKE:
                ChargingHandshakeMessage chargingHandshakeMessage = JSON.parseObject(jsonObject.toJSONString(),ChargingHandshakeMessage.class);
                ChargingHandshakeMessage chargingHandshakeMessage = JSON.parseObject(content.toJSONString(),ChargingHandshakeMessage.class);
                sendResult = enhanceProduce.chargingHandshakeMessage(chargingHandshakeMessage);
                // 响应硬件
                break;
            case SendTagConstant.PARAMETER_SETTING:
                ParameterSettingMessage parameterSettingMessage = JSON.parseObject(jsonObject.toJSONString(),ParameterSettingMessage.class);
                ParameterSettingMessage parameterSettingMessage = JSON.parseObject(content.toJSONString(),ParameterSettingMessage.class);
                sendResult = enhanceProduce.parameterSettingMessage(parameterSettingMessage);
                break;
            case SendTagConstant.BMS_ABORT:
                BmsAbortMessage bmsAbortMessage = JSON.parseObject(jsonObject.toJSONString(),BmsAbortMessage.class);
                BmsAbortMessage bmsAbortMessage = JSON.parseObject(content.toJSONString(),BmsAbortMessage.class);
                sendResult = enhanceProduce.bmsAbortMessage(bmsAbortMessage);
                // 响应硬件
                break;
            case SendTagConstant.MOTOR_ABORT:
                MotorAbortMessage motorAbortMessage = JSON.parseObject(jsonObject.toJSONString(),MotorAbortMessage.class);
                MotorAbortMessage motorAbortMessage = JSON.parseObject(content.toJSONString(),MotorAbortMessage.class);
                sendResult = enhanceProduce.motorAbortMessage(motorAbortMessage);
                break;
            case SendTagConstant.BMS_DEMAND_AND_CHARGER_EXPORTATION:
                BmsDemandAndChargerExportationMessage bmsDemandAndChargerExportationMessage = JSON.parseObject(jsonObject.toJSONString(),BmsDemandAndChargerExportationMessage.class);
                BmsDemandAndChargerExportationMessage bmsDemandAndChargerExportationMessage = JSON.parseObject(content.toJSONString(),BmsDemandAndChargerExportationMessage.class);
                sendResult = enhanceProduce.bmsDemandAndChargerExportationMessage(bmsDemandAndChargerExportationMessage);
                // 响应硬件
                break;
            case SendTagConstant.BMS_INFORMATION:
                BmsInformationMessage bmsInformationMessage = JSON.parseObject(jsonObject.toJSONString(),BmsInformationMessage.class);
                BmsInformationMessage bmsInformationMessage = JSON.parseObject(content.toJSONString(),BmsInformationMessage.class);
                sendResult = enhanceProduce.bmsInformationMessage(bmsInformationMessage);
                // 响应硬件
                break;
            case SendTagConstant.CHARGING_PILE_STARTS_CHARGING:
                ChargingPileStartsChargingMessage chargingPileStartsChargingMessage = JSON.parseObject(jsonObject.toJSONString(),ChargingPileStartsChargingMessage.class);
                ChargingPileStartsChargingMessage chargingPileStartsChargingMessage = JSON.parseObject(content.toJSONString(),ChargingPileStartsChargingMessage.class);
                sendResult = enhanceProduce.chargingPileStartsChargingMessage(chargingPileStartsChargingMessage);
                // 响应硬件
                break;
            case SendTagConstant.PLATFORM_START_CHARGING_REPLY:
                PlatformStartChargingReplyMessage platformStartChargingReplyMessage = JSON.parseObject(jsonObject.toJSONString(),PlatformStartChargingReplyMessage.class);
                PlatformStartChargingReplyMessage platformStartChargingReplyMessage = JSON.parseObject(content.toJSONString(),PlatformStartChargingReplyMessage.class);
                sendResult = enhanceProduce.platformStartChargingReplyMessage(platformStartChargingReplyMessage);
                // 响应硬件
                break;
            case SendTagConstant.PLATFORM_STOP_CHARGING_REPLY:
                PlatformStopChargingReplyMessage platformStopChargingReplyMessage = JSON.parseObject(jsonObject.toJSONString(),PlatformStopChargingReplyMessage.class);
                PlatformStopChargingReplyMessage platformStopChargingReplyMessage = JSON.parseObject(content.toJSONString(),PlatformStopChargingReplyMessage.class);
                sendResult = enhanceProduce.platformStopChargingReplyMessage(platformStopChargingReplyMessage);
                // 响应硬件
                break;
            case SendTagConstant.TRANSACTION_RECORD:
                TransactionRecordMessage transactionRecordMessage = JSON.parseObject(jsonObject.toJSONString(),TransactionRecordMessage.class);
                TransactionRecordMessage transactionRecordMessage = JSON.parseObject(content.toJSONString(),TransactionRecordMessage.class);
                sendResult = enhanceProduce.transactionRecordMessage(transactionRecordMessage);
                // 响应硬件
                ConfirmTransactionRecord confirmTransactionRecord = new ConfirmTransactionRecord();
@@ -187,27 +187,27 @@
                iotMessageProduce.sendMessage(deviceId, ServiceIdMenu.CONFIRM_TRANSACTION_RECORD.getKey(),messageUtil.confirmTransactionRecord(confirmTransactionRecord));
                break;
            case SendTagConstant.UPDATE_BALANCE_REPLY:
                UpdateBalanceReplyMessage updateBalanceReplyMessage = JSON.parseObject(jsonObject.toJSONString(),UpdateBalanceReplyMessage.class);
                UpdateBalanceReplyMessage updateBalanceReplyMessage = JSON.parseObject(content.toJSONString(),UpdateBalanceReplyMessage.class);
                sendResult = enhanceProduce.updateBalanceReplyMessage(updateBalanceReplyMessage);
                // 响应硬件
                break;
            case SendTagConstant.SYNCHRONIZE_OFFLINE_CARD_REPLY:
                SynchronizeOfflineCardReplyMessage synchronizeOfflineCardReplyMessage = JSON.parseObject(jsonObject.toJSONString(),SynchronizeOfflineCardReplyMessage.class);
                SynchronizeOfflineCardReplyMessage synchronizeOfflineCardReplyMessage = JSON.parseObject(content.toJSONString(),SynchronizeOfflineCardReplyMessage.class);
                sendResult = enhanceProduce.synchronizeOfflineCardReplyMessage(synchronizeOfflineCardReplyMessage);
                // 响应硬件
                break;
            case SendTagConstant.CLEAR_OFFLINE_CARD_REPLY:
                ClearOfflineCardReplyMessage clearOfflineCardReplyMessage = JSON.parseObject(jsonObject.toJSONString(),ClearOfflineCardReplyMessage.class);
                ClearOfflineCardReplyMessage clearOfflineCardReplyMessage = JSON.parseObject(content.toJSONString(),ClearOfflineCardReplyMessage.class);
                sendResult = enhanceProduce.clearOfflineCardReplyMessage(clearOfflineCardReplyMessage);
                // 响应硬件
                break;
            case SendTagConstant.WORKING_PARAMETER_SETTING_REPLY:
                WorkingParameterSettingReplyMessage workingParameterSettingReplyMessage = JSON.parseObject(jsonObject.toJSONString(),WorkingParameterSettingReplyMessage.class);
                WorkingParameterSettingReplyMessage workingParameterSettingReplyMessage = JSON.parseObject(content.toJSONString(),WorkingParameterSettingReplyMessage.class);
                sendResult = enhanceProduce.workingParameterSettingReplyMessage(workingParameterSettingReplyMessage);
                // 响应硬件
                break;
            case SendTagConstant.TIMING_SETTING:
                TimingSettingMessage timingSettingMessage = JSON.parseObject(jsonObject.toJSONString(),TimingSettingMessage.class);
                TimingSettingMessage timingSettingMessage = JSON.parseObject(content.toJSONString(),TimingSettingMessage.class);
                sendResult = enhanceProduce.timingSettingMessage(timingSettingMessage);
                // 响应硬件 对时设置应答
                TimingSettingReply timingSettingReply = new TimingSettingReply();
@@ -216,27 +216,27 @@
                iotMessageProduce.sendMessage(deviceId, ServiceIdMenu.TIMING_SETTING_REPLY.getKey(),messageUtil.timingSettingReply(timingSettingReply));
                break;
            case SendTagConstant.SETUP_BILLING_MODEL_REPLY:
                SetupBillingModelReplyMessage setupBillingModelReplyMessage = JSON.parseObject(jsonObject.toJSONString(),SetupBillingModelReplyMessage.class);
                SetupBillingModelReplyMessage setupBillingModelReplyMessage = JSON.parseObject(content.toJSONString(),SetupBillingModelReplyMessage.class);
                sendResult = enhanceProduce.setupBillingModelReplyMessage(setupBillingModelReplyMessage);
                // 响应硬件
                break;
            case SendTagConstant.GROUND_LOCK_REAL_TIME_DATA:
                GroundLockRealTimeDataMessage groundLockRealTimeDataMessage = JSON.parseObject(jsonObject.toJSONString(),GroundLockRealTimeDataMessage.class);
                GroundLockRealTimeDataMessage groundLockRealTimeDataMessage = JSON.parseObject(content.toJSONString(),GroundLockRealTimeDataMessage.class);
                sendResult = enhanceProduce.groundLockRealTimeDataMessage(groundLockRealTimeDataMessage);
                // 响应硬件
                break;
            case SendTagConstant.CHARGING_PILE_RETURNS_GROUND_LOCK_DATA:
                ChargingPileReturnsGroundLockDataMessage chargingPileReturnsGroundLockDataMessage = JSON.parseObject(jsonObject.toJSONString(),ChargingPileReturnsGroundLockDataMessage.class);
                ChargingPileReturnsGroundLockDataMessage chargingPileReturnsGroundLockDataMessage = JSON.parseObject(content.toJSONString(),ChargingPileReturnsGroundLockDataMessage.class);
                sendResult = enhanceProduce.chargingPileReturnsGroundLockDataMessage(chargingPileReturnsGroundLockDataMessage);
                // 响应硬件
                break;
            case SendTagConstant.PLATFORM_RESTART_REPLY:
                PlatformRestartReplyMessage platformRestartReplyMessage = JSON.parseObject(jsonObject.toJSONString(),PlatformRestartReplyMessage.class);
                PlatformRestartReplyMessage platformRestartReplyMessage = JSON.parseObject(content.toJSONString(),PlatformRestartReplyMessage.class);
                sendResult = enhanceProduce.platformRestartReplyMessage(platformRestartReplyMessage);
                // 响应硬件
                break;
            default:
                PlatformRemoteUpdateReplyMessage platformRemoteUpdateReplyMessage = JSON.parseObject(jsonObject.toJSONString(),PlatformRemoteUpdateReplyMessage.class);
                PlatformRemoteUpdateReplyMessage platformRemoteUpdateReplyMessage = JSON.parseObject(content.toJSONString(),PlatformRemoteUpdateReplyMessage.class);
                sendResult = enhanceProduce.platformRemoteUpdateReplyMessage(platformRemoteUpdateReplyMessage);
                // 响应硬件
                break;

 ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/rocket/listener/OnlineMessageListener.java

@@ -1,16 +1,10 @@
package com.ruoyi.integration.rocket.listener;

import com.ruoyi.integration.api.model.Online;
import com.ruoyi.integration.api.model.OnlineReply;
import com.ruoyi.integration.iotda.enums.ServiceIdMenu;
import com.ruoyi.integration.iotda.utils.produce.IotMessageProduce;
import com.ruoyi.integration.iotda.utils.tools.MessageUtil;
import com.ruoyi.integration.mongodb.service.OnlineService;
import com.ruoyi.integration.rocket.model.OnlineMessage;
import com.ruoyi.integration.rocket.model.OnlineMessage;
import com.ruoyi.integration.rocket.util.EnhanceMessageHandler;
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;
@@ -20,7 +14,6 @@
@Slf4j
@Component
@RocketMQMessageListener(
        messageModel = MessageModel.CLUSTERING,
        consumerGroup = "charge_online",
        topic = "charge_online",
        selectorExpression = "online", // 明确指定标签

 ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/rocket/listener/PingMessageListener.java

@@ -1,16 +1,12 @@
package com.ruoyi.integration.rocket.listener;

import com.ruoyi.integration.api.model.Online;
import com.ruoyi.integration.api.model.Ping;
import com.ruoyi.integration.api.model.Pong;
import com.ruoyi.integration.iotda.enums.ServiceIdMenu;
import com.ruoyi.integration.iotda.utils.produce.IotMessageProduce;
import com.ruoyi.integration.iotda.utils.tools.MessageUtil;
import com.ruoyi.integration.mongodb.service.OnlineService;
import com.ruoyi.integration.mongodb.service.PingService;
import com.ruoyi.integration.rocket.model.OnlineMessage;
import com.ruoyi.integration.rocket.model.PingMessage;
import com.ruoyi.integration.rocket.util.EnhanceMessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;

 ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/rocket/listener/TransactionRecordMessageListener.java

@@ -9,6 +9,8 @@
import com.ruoyi.integration.mongodb.service.TransactionRecordService;
import com.ruoyi.integration.rocket.model.TransactionRecordMessage;
import com.ruoyi.integration.rocket.util.EnhanceMessageHandler;
import com.ruoyi.order.api.feignClient.ChargingOrderClient;
import com.ruoyi.order.api.model.TChargingOrder;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
@@ -16,6 +18,8 @@
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Objects;

@Slf4j
@Component
@@ -31,9 +35,7 @@
    @Autowired
    private TransactionRecordService transactionRecordService;
    @Autowired
    private IotMessageProduce iotMessageProduce;
    @Autowired
    private MessageUtil messageUtil;
    private ChargingOrderClient chargingOrderClient;
    @Override
    protected void handleMessage(TransactionRecordMessage message) throws Exception {
        // 此时这里才是最终的业务处理，代码只需要处理资源类关闭异常，其他的可以交给父类重试
@@ -43,6 +45,11 @@
        BeanUtils.copyProperties(message,transactionRecord);
        transactionRecordService.create(transactionRecord);
        // 业务处理
        TChargingOrder chargingOrder = chargingOrderClient.getOrderByCode(message.getTransaction_serial_number()).getData();
        if(Objects.nonNull(chargingOrder)){
            chargingOrder.setTotalElectricity(message.getTotal_electricity());
            chargingOrderClient.updateChargingOrder(chargingOrder);
        }
    }

    @Override

 ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/rocket/util/RocketMQEnhanceTemplate.java

@@ -5,10 +5,12 @@
import com.ruoyi.integration.rocket.configuration.RocketEnhanceProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.StringUtils;
@@ -24,9 +26,6 @@
    private RocketEnhanceProperties rocketEnhanceProperties;

    public RocketMQTemplate getTemplate() {
//        DefaultMQProducer producer = new DefaultMQProducer();
//        producer.setProducerGroup("enhance_consumer_group");
//        template.setProducer(producer);
        return template;
    }

@@ -62,7 +61,7 @@
    public <T extends BaseMessage> SendResult send(String destination, T message) {
        // 设置业务键，此处根据公共的参数进行处理
        // 更多的其它基础业务处理...
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.TAGS, message.getKey()).build();
        SendResult sendResult = template.syncSend(destination, sendMessage);
        // 此处为了方便查看给日志转了json，根据选择选择日志记录方式，例如ELK采集
        log.info("[{}]同步消息[{}]发送结果[{}]", destination, JSONObject.toJSON(message), JSONObject.toJSON(sendResult));
@@ -77,7 +76,7 @@
    }

    public <T extends BaseMessage> SendResult send(String destination, T message, int delayLevel) {
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.TAGS, message.getKey()).build();
        SendResult sendResult = template.syncSend(destination, sendMessage, 3000, delayLevel);
        log.info("[{}]延迟等级[{}]消息[{}]发送结果[{}]", destination, delayLevel, JSONObject.toJSON(message), JSONObject.toJSON(sendResult));
        return sendResult;

 ruoyi-service/ruoyi-order/src/main/resources/mapper/order/TChargingOrderMapper.xml

@@ -39,6 +39,7 @@
        <result column="need_elec" property="needElec" />
        <result column="create_time" property="createTime" />
        <result column="del_flag" property="delFlag" />
        <result column="total_electricity" property="totalElectricity" />
    </resultMap>

    <!-- 通用查询结果列 -->
@@ -46,7 +47,7 @@
        id, code, order_type, order_classification, app_user_id, app_user_car_id,site_id, parking_lot_id, charging_pile_id, charging_gun_id,
            start_time, end_time, status, recharge_amount, recharge_payment_type, recharge_payment_status, recharge_serial_number,
            order_amount, app_coupon_id, coupon_discount_amount, vip_discount, vip_discount_amount, payment_amount, refund_code,
            refund_amount, refund_status, refund_serial_number, refund_time, create_time, del_flag,need_elec
            refund_amount, refund_status, refund_serial_number, refund_time, create_time, del_flag,need_elec,total_electricity
    </sql>