~opt:queue.order.close;

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/AbstractOrderCloseReceiver.java

@@ -22,9 +22,11 @@ import org.springframework.data.redis.connection.stream.StreamRecords;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.stream.StreamListener;
 
+import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.HashMap;
 import java.util.LinkedList;
+import java.util.Map;
 
 import static com.liquidnet.commons.lang.util.DateUtil.DTF_YMD_HMS;
 
@@ -41,10 +43,10 @@ public abstract class AbstractOrderCloseReceiver implements StreamListener<Strin
 
     @Override
     public void onMessage(MapRecord<String, String, String> message) {
-        log.debug("CONSUMER XLS [streamKey:{},messageId:{},stream:{},body:{}]",
+        log.debug("CONSUMER ORDER_CLOSE [streamKey:{},messageId:{},stream:{},body:{}]",
                 this.getRedisStreamKey(), message.getId(), message.getStream(), message.getValue());
-        boolean result = this.consumerOrderCloseHandler(this.getRedisStreamKey(), message.getValue().get("message"), message.getValue().get("type"));
-        log.info("CONSUMER XLS_PATH RESULT:{} ==> MESSAGE_ID:{}", result, message.getId());
+        boolean result = this.consumerOrderCloseHandler(message.getValue());
+        log.info("CONSUMER ORDER_CLOSE RESULT:{} ==> MESSAGE_ID:{}", result, message.getId());
 
         try {
             stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
@@ -60,44 +62,33 @@ public abstract class AbstractOrderCloseReceiver implements StreamListener<Strin
         }
     }
 
-    private int getGoblinOrderUnPayNum(String streamKey) {
-        String[] keyArr = streamKey.split(":");
-        return Integer.parseInt(keyArr[keyArr.length]);
-    }
-
-    private boolean consumerOrderCloseHandler(String streamKey, String message, String type) {
-        boolean aBoolean = false;
+    private boolean consumerOrderCloseHandler(Map<String, String> messageMap) {
         try {
-            int unPayNum = this.getGoblinOrderUnPayNum(streamKey);
-            LocalDateTime now = LocalDateTime.now().minusMinutes(5);
-            int currMin = now.getMinute() % 10, sleepMin = 1;
-            if (currMin == unPayNum) {
-                aBoolean = checkOrderTime(message, type);
+            String orderCode = messageMap.get("id"), type = messageMap.get("type");
+            LocalDateTime createdAt = LocalDateTime.parse(messageMap.get("time"));
+
+            long durationToMillis = Duration.between(createdAt.plusMinutes(5), LocalDateTime.now()).toMillis();
+
+            if (durationToMillis > 0) {
+                return checkOrderTime(orderCode, type);
             } else {
                 try {
-                    Thread.sleep(sleepMin * 60000L);
-                    consumerOrderCloseHandler(streamKey, message, type);
+                    Thread.sleep(Math.abs(durationToMillis));
                 } catch (InterruptedException ignored) {
                 }
+                return consumerOrderCloseHandler(messageMap);
             }
         } catch (Exception e) {
-            log.error("CONSUMER ORDER {} CLOSE FAIL ==> {}", type, e.getMessage(), e);
-        } finally {
-            if (!aBoolean) {
-                HashMap<String, String> map = CollectionUtil.mapStringString();
-                map.put("message", message);
-                map.put("type", type);
-                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
-            }
+            log.error("CONSUMER ORDER_CLOSE FAIL ==> {}", e.getMessage(), e);
+            stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(messageMap).withStreamKey(this.getRedisStreamKey()));
+            return false;
         }
-        return aBoolean;
     }
 
     protected abstract String getRedisStreamKey();
 
     protected abstract String getRedisStreamGroup();
 
-
     public boolean checkOrderTime(String valueData, String type) {
         LocalDateTime now = LocalDateTime.now();
         if (type.equals("GOBLIN")) {

liquidnet-bus-service/liquidnet-service-order/liquidnet-service-order-impl/src/main/java/com/liquidnet/service/order/utils/QueueUtils.java

 package com.liquidnet.service.order.utils;
 
-import com.liquidnet.commons.lang.util.RandomUtil;
 import com.liquidnet.service.base.SqlMapping;
 import com.liquidnet.service.base.constant.MQConst;
 import com.liquidnet.service.kylin.dto.vo.mongo.KylinIpAreaVo;
@@ -87,9 +86,10 @@ public class QueueUtils {
                 streamKey = MQConst.GoblinQueue.GOBLIN_UN_PAY_0.getKey();
                 break;
         }
-        HashMap<String, String> map = ObjectUtil.cloneHashMapStringAndString();
-        map.put("message", masterOrderCode);
+        HashMap<String, Object> map = ObjectUtil.cloneHashMapStringAndObject();
+        map.put("id", masterOrderCode);
         map.put("type", "GOBLIN");
+        map.put("time", createTime.toString());
         stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(streamKey));
     }