bug修改 供应商的下单扣款和失败退款操作并记录操作流水

modules/callback-customer/src/main/java/com/fire/customer/callback/consumer/RocketUpdateConsumer.java

@@ -0,0 +1,41 @@
+package com.fire.customer.callback.consumer;
+
+import com.fire.customer.callback.service.CacheService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+import static com.fire.common.constants.RocketTags.CONSUMER_PRODUCT_TAG;
+import static com.fire.common.constants.RocketTopic.UPDATE_TOPIC;
+
+
+/**
+ * 消息消费者，可与生产者分离
+ * 这里定义的messageMode是集群消费，如果是同一消费组则会组内消费者均衡消费；
+ * 如果是不同消费组，则会起到广播消费的效果
+ *
+ * @author ZJQ 2021年5月26日14:59:42
+ */
+
+@Slf4j
+@Component
+@RocketMQMessageListener(consumerGroup = "${rocketmq.consumer.group-cache}", topic = UPDATE_TOPIC, selectorExpression = CONSUMER_PRODUCT_TAG)
+public class RocketUpdateConsumer implements RocketMQListener<MessageExt> {
+
+    @Resource
+    private CacheService cacheService;
+
+    @Override
+    public void onMessage(MessageExt msg) {
+        //只消费最近十分钟的消息，因为再之前的消息为历史的更新没有意义
+        Long now = System.currentTimeMillis();
+        Long bron = msg.getBornTimestamp();
+        if (now - bron < 600000) {
+            cacheService.cacheCustomer();
+        }
+    }
+}

modules/callback-customer/src/main/java/com/fire/customer/callback/service/impl/CallbackCustomerServiceImpl.java

@@ -50,6 +50,8 @@ public class CallbackCustomerServiceImpl implements CallbackCustomerService {
         content.setCheckTime(oldTimestamp);
         content.setOrderId(orderInfo.getOrderId());
         content.setPrice(orderInfo.getPrice());
+        content.setCode(String.valueOf(orderInfo.getStatus()));
+        content.setStatus(orderInfo.getNote());
         //设置header中的参数
         header.setTimestamp(oldTimestamp);
         header.setAppId(orderInfo.getAppId());

modules/callback-customer/src/main/java/com/fire/customer/callback/task/CacheCustomerTask.java

@@ -0,0 +1,43 @@
+package com.fire.customer.callback.task;
+
+
+import com.fire.customer.callback.service.CacheService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
+
+/**
+ * 缓存任务
+ *
+ * @author ZJQ 2021年5月18日13:41:01
+ */
+@Component
+@Slf4j
+public class CacheCustomerTask {
+    @Resource
+    private CacheService cacheService;
+
+    /**
+     * 每天凌晨2时5分0秒定时缓存客户信息
+     */
+    @Async
+    @Scheduled(cron = "0 5 2 * * ?")
+    public void cacheCustomer() {
+        cacheService.cacheCustomer();
+    }
+
+    /**
+     * spring启动时执行缓存初始化客户信息
+     */
+    @Async
+    @PostConstruct
+    public void initCustomer() {
+        cacheService.cacheCustomer();
+        log.info("spring启动时执行缓存初始化客户信息");
+    }
+
+}

modules/callback-customer/src/main/resources/bootstrap.yml

@@ -4,7 +4,7 @@ spring:
   application:
     name: callback-customer
   profiles:
-    active: dev
+    active: test
 ---
 spring:
   config:

modules/distribution/src/main/resources/bootstrap.yml

@@ -4,7 +4,7 @@ spring:
   application:
     name: distribution
   profiles:
-    active: dev
+    active: test
 ---
 spring:
   config:

modules/get-order-supplier/src/main/java/com/fire/get/order/service/impl/GetOrderSupplierServiceImpl.java

@@ -4,10 +4,12 @@ package com.fire.get.order.service.impl;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fire.common.auth.SupplierAuth;
+import com.fire.common.redis.RedisAmountScript;
 import com.fire.common.redis.RedisPriorityQueueScript;
 import com.fire.dto.ChannelSupplier;
 import com.fire.dto.FlowOrderInfo;
 import com.fire.dto.MobileFlowDispatchRec;
+import com.fire.dto.TransactionFlow;
 import com.fire.dto.enums.BusinessStatus;
 import com.fire.dto.enums.SupplierOrderStatus;
 import com.fire.get.order.data.DataPool;
@@ -22,19 +24,19 @@ import org.springframework.web.client.RestTemplate;
 import redis.clients.jedis.JedisCluster;
 
 import javax.annotation.Resource;
-
 import java.util.Date;
 
 import static com.fire.common.constants.RocketTags.FAIL_TAG;
 import static com.fire.common.constants.RocketTags.SUCCESS_TAG;
-import static com.fire.common.constants.RocketTopic.CHILD_ORDER_TOPIC;
-import static com.fire.common.constants.RocketTopic.ORDER_TOPIC;
+import static com.fire.common.constants.RocketTopic.*;
+import static com.fire.dto.enums.AmountOper.ADD;
 import static com.fire.dto.enums.BusinessStatus.BUSINESS_REPORT_ERROR;
 import static com.fire.dto.enums.BusinessStatus.BUSINESS_SUCCESS;
 import static com.fire.dto.enums.OrderStatus.ORDER_FAIL;
 import static com.fire.dto.enums.OrderStatus.ORDER_SUCCESS;
-import static com.fire.dto.enums.RedisKey.CHILD_ORDER_INFO;
-import static com.fire.dto.enums.RedisKey.ORDER_INFO;
+import static com.fire.dto.enums.RedisKey.*;
+import static com.fire.dto.enums.RelationType.RELATION_SUPPLIER;
+import static com.fire.dto.enums.ServiceType.SERVICE_ORDER_ADD;
 import static com.fire.dto.enums.SupplierOrderStatus.SUPPLIER_ORDER_PROCESS;
 
 /**
@@ -56,10 +58,13 @@ public class GetOrderSupplierServiceImpl implements GetOrderSupplierService {
     private RocketMQTemplate rocketMQTemplate;
     @Resource
     private SupplierAuth supplierAuth;
+    @Resource
+    private RedisAmountScript redisAmountScript;
 
     @Override
     @Async
     public void getOrder(MobileFlowDispatchRec childOrder) {
+
         //根据通道提取供应商信息
         ChannelSupplier supplier = DataPool.supplierHashMap.get(childOrder.getChannelId());
         //组装请求参数
@@ -105,27 +110,71 @@ public class GetOrderSupplierServiceImpl implements GetOrderSupplierService {
 
         //如果充值中不操作 所以非充值中都需要更改状态
         if (!supplierOrderStatus.equals(SUPPLIER_ORDER_PROCESS)) {
-            //分发记录修改 只要不是充值中，子订单就进入终态 todo redis中获取分发记录
+            //分发记录修改 只要不是充值中，子订单就进入终态
+            ObjectMapper om = new ObjectMapper();
+            //获取一个实时的子订单状态
+            Long recId = childOrder.getRecId();
+            long hKey = recId / 10000000;
+
+            try {
+                String newChildOrderStr = jedisCluster.hget(CHILD_ORDER_INFO.key() + hKey, String.valueOf(recId));
+                MobileFlowDispatchRec newChildOrder = om.readValue(newChildOrderStr, MobileFlowDispatchRec.class);
+                //如果当前是最终态 直接返回
+                if (ORDER_FAIL.status().equals(newChildOrder.getSendStatus()) || ORDER_SUCCESS.status().equals(newChildOrder.getSendStatus())) {
+                    return;
+                }
+            } catch (Exception e) {
+                log.error("实时子订单信息获取异常：", e);
+            }
+
 
             childOrder.setGwErrorMsg(businessStatus.message());
             childOrder.setGwErrorCode(String.valueOf(businessStatus.status()));
+            String tags = null;
             switch (supplierOrderStatus) {
-                case SUPPLIER_ORDER_FAILED -> childOrder.setSendStatus(ORDER_FAIL.status());
-                case SUPPLIER_ORDER_SUCCESS -> childOrder.setSendStatus(ORDER_SUCCESS.status());
+                case SUPPLIER_ORDER_FAILED -> {
+                    childOrder.setSendStatus(ORDER_FAIL.status());
+                    tags = FAIL_TAG;
+                    //子订单失败 如果原来已经是最终态
+                    Long afterAmount = redisAmountScript.changeAmount(SUPPLIER_AMOUNT.key(), supplier.getSupplierId(), ADD.oper(), childOrder.getOperatorBalancePrice(), 100000000000L);
+                    Long transSeq = jedisCluster.incr(GLOBAL_ID_INCR.key());
+                    TransactionFlow transaction = TransactionFlow.builder()
+                            .afterAmount(afterAmount)
+                            .beforeAmount(afterAmount + childOrder.getOperatorBalancePrice())
+                            .createTime(new Date())
+                            .distinguish(RELATION_SUPPLIER.type())
+                            .extorderId(String.valueOf(childOrder.getRecId()))
+                            .orderId(childOrder.getOrderId())
+                            .name(supplier.getSupplierName())
+                            .note("订单失败退款")
+                            .operatorName("system")
+                            .operatingAmount(childOrder.getOperatorBalancePrice())
+                            .serviceType(SERVICE_ORDER_ADD.type())
+                            .relationId(supplier.getSupplierId())
+                            .seqNo(transSeq)
+                            .build();
+                    try {
+                        //流水入队列
+                        String transactionStr = om.writeValueAsString(transaction);
+                        rocketMQTemplate.syncSend(TRANSACTION_TOPIC, MessageBuilder.withPayload(transactionStr).build());
+                    } catch (Exception e) {
+                        log.error("供应商下单扣减异常：", e);
+                    }
+                }
+                case SUPPLIER_ORDER_SUCCESS -> {
+                    childOrder.setSendStatus(ORDER_SUCCESS.status());
+                    tags = SUCCESS_TAG;
+                }
             }
-            Long recId = childOrder.getRecId();
+
             String childOrderStr;
-            ObjectMapper om = new ObjectMapper();
+
             try {
                 childOrderStr = om.writeValueAsString(childOrder);
                 //订单转换为json更新入redis
-                long hKey = recId / 10000000;
+
                 jedisCluster.hset(CHILD_ORDER_INFO.key() + hKey, String.valueOf(recId), childOrderStr);
-                String tags = null;
-                switch (supplierOrderStatus) {
-                    case SUPPLIER_ORDER_FAILED -> tags = FAIL_TAG;
-                    case SUPPLIER_ORDER_SUCCESS -> tags = SUCCESS_TAG;
-                }
+
                 //子订单入队列
                 rocketMQTemplate.syncSendOrderly(CHILD_ORDER_TOPIC + ":" + tags, MessageBuilder.withPayload(childOrderStr).build(), String.valueOf(recId));
             } catch (JsonProcessingException e) {
@@ -136,9 +185,9 @@ public class GetOrderSupplierServiceImpl implements GetOrderSupplierService {
 
             long orderId = childOrder.getOrderId();
             //订单转换为json更新入redis
-            long hKey = orderId / 10000000;
             String orderStr = jedisCluster.hget(ORDER_INFO.key() + hKey, String.valueOf(orderId));
 
+
             //子订单成功  那么主订单就成功了 子订单成功 那么主订单重入队列
             switch (supplierOrderStatus) {
                 case SUPPLIER_ORDER_SUCCESS -> {

modules/get-order-supplier/src/main/resources/bootstrap.yml

@@ -4,7 +4,7 @@ spring:
   application:
     name: get-order-supplier
   profiles:
-    active: dev
+    active: test
 ---
 spring:
   config:

modules/make-order-supplier/src/main/java/com/fire/order/supplier/service/impl/MakeOrderSupplierServiceImpl.java

@@ -4,31 +4,43 @@ package com.fire.order.supplier.service.impl;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fire.common.auth.SupplierAuth;
+import com.fire.common.redis.RedisAmountScript;
 import com.fire.common.redis.RedisPriorityQueueScript;
 import com.fire.dto.ChannelSupplier;
 import com.fire.dto.FlowOrderInfo;
 import com.fire.dto.MobileFlowDispatchRec;
+import com.fire.dto.TransactionFlow;
 import com.fire.dto.enums.BusinessStatus;
 import com.fire.order.supplier.data.DataPool;
 import com.fire.order.supplier.service.MakeOrderSupplierService;
 import com.fire.supplier.dto.*;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.http.ResponseEntity;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.RestTemplate;
 import redis.clients.jedis.JedisCluster;
 
 import javax.annotation.Resource;
+import java.io.IOException;
 import java.math.BigDecimal;
+import java.util.Date;
 
 import static com.fire.common.constants.RocketTags.FAIL_TAG;
 import static com.fire.common.constants.RocketTopic.CHILD_ORDER_TOPIC;
+import static com.fire.common.constants.RocketTopic.TRANSACTION_TOPIC;
+import static com.fire.dto.enums.AmountOper.SUB;
 import static com.fire.dto.enums.BusinessStatus.BUSINESS_REPORT_ERROR;
 import static com.fire.dto.enums.BusinessStatus.BUSINESS_SUCCESS;
 import static com.fire.dto.enums.OrderStatus.ORDER_FAIL;
-import static com.fire.dto.enums.RedisKey.CHILD_ORDER_INFO;
+import static com.fire.dto.enums.RedisKey.*;
+import static com.fire.dto.enums.RelationType.RELATION_SUPPLIER;
+import static com.fire.dto.enums.ServiceType.SERVICE_ORDER_SUB;
 
 /**
  * 向供应商下单实现
@@ -49,6 +61,8 @@ public class MakeOrderSupplierServiceImpl implements MakeOrderSupplierService {
     private RocketMQTemplate rocketMQTemplate;
     @Resource
     private SupplierAuth supplierAuth;
+    @Resource
+    private RedisAmountScript redisAmountScript;
 
     @Override
     public void makeOrder(FlowOrderInfo orderInfo) {
@@ -114,10 +128,12 @@ public class MakeOrderSupplierServiceImpl implements MakeOrderSupplierService {
             //失败直接重入优先级队列 同时更新分发的状态
             businessStatus = BUSINESS_REPORT_ERROR;
         }
+        //分发记录修改
+        MobileFlowDispatchRec newChildOrder = orderInfo.getNewChildOrder();
+
         //只要不成功那么重入优先级队列并且更新分发状态
         if (!BUSINESS_SUCCESS.equals(businessStatus)) {
-            //分发记录修改
-            MobileFlowDispatchRec newChildOrder = orderInfo.getNewChildOrder();
+
             newChildOrder.setGwErrorMsg(BUSINESS_SUCCESS.message());
             newChildOrder.setGwErrorCode(String.valueOf(BUSINESS_SUCCESS.status()));
             newChildOrder.setSendStatus(ORDER_FAIL.status());
@@ -144,6 +160,32 @@ public class MakeOrderSupplierServiceImpl implements MakeOrderSupplierService {
                 log.error("订单重入优先级队列失败：", e);
                 //todo 进入兜底订单
             }
+        } else {
+            //下单成功扣减
+            Long afterAmount = redisAmountScript.changeAmount(SUPPLIER_AMOUNT.key(), supplier.getSupplierId(), SUB.oper(), newChildOrder.getOperatorBalancePrice(), 100000000000L);
+            Long transSeq = jedisCluster.incr(GLOBAL_ID_INCR.key());
+            TransactionFlow transaction = TransactionFlow.builder()
+                    .afterAmount(afterAmount)
+                    .beforeAmount(afterAmount + newChildOrder.getOperatorBalancePrice())
+                    .createTime(new Date())
+                    .distinguish(RELATION_SUPPLIER.type())
+                    .extorderId(String.valueOf(newChildOrder.getRecId()))
+                    .orderId(newChildOrder.getOrderId())
+                    .name(supplier.getSupplierName())
+                    .note("订单充值扣减")
+                    .operatorName("system")
+                    .operatingAmount(newChildOrder.getOperatorBalancePrice())
+                    .serviceType(SERVICE_ORDER_SUB.type())
+                    .relationId(supplier.getSupplierId())
+                    .seqNo(transSeq)
+                    .build();
+            try {
+                //流水入队列
+                String transactionStr = om.writeValueAsString(transaction);
+                rocketMQTemplate.syncSend(TRANSACTION_TOPIC, MessageBuilder.withPayload(transactionStr).build());
+            } catch (Exception e) {
+                log.error("供应商下单扣减异常：", e);
+            }
         }
     }
 

modules/make-order/src/main/java/com/fire/order/service/impl/MakeOrderServiceImpl.java

@@ -370,6 +370,7 @@ public class MakeOrderServiceImpl implements MakeOrderService {
                 .orderId(orderId)
                 .name(customerInfo.getCustomerName())
                 .note("订单充值扣减")
+                .operatorName("system")
                 .operatingAmount(product.getPrice())
                 .serviceType(SERVICE_ORDER_SUB.type())
                 .relationId(flowAppInfo.getCustomerId())