+queue.tomdb;

liquidnet-bus-api/liquidnet-service-goblin-api/src/main/java/com/liquidnet/service/goblin/dto/GoblinQueueBizMongoDto.java

@@ -14,19 +14,26 @@ import java.io.Serializable;
 public class GoblinQueueBizMongoDto implements Serializable, Cloneable {
 
     private static final long serialVersionUID = -5850588358672254766L;
-
     /**
-     * mongo操作入队列
-     *
-     * @param bizId 业务id 例如orderId
-     * @param bizType 业务类型 1NFT订单 2...
-     * @param opType 操作类型 1创建 2修改
-     * @param listType 队列类型 1即时性要求高队列 2即时性要求不高队列
+     * 业务类型[]
+     */
+    private String bizType;
+    /**
+     * Mongo集合中的字段属性名称
+     */
+    private String column;
+    /**
+     * Mongo集合中的字段属性值
      */
     private String bizId;
-    private int bizType;
+    /**
+     * Redis缓存中的Key前缀
+     */
+    private String prefix;
+    /**
+     * 操作类型[1-insert|2-update]
+     */
     private int opType;
-    private int listType;
 
     private static final GoblinQueueBizMongoDto obj = new GoblinQueueBizMongoDto();
 

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

 package com.liquidnet.service.consumer.kylin.receiver;
 
+import com.liquidnet.common.cache.redis.util.RedisUtil;
 import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.commons.lang.util.JsonUtils;
-import com.liquidnet.service.base.SqlMapping;
 import com.liquidnet.service.base.constant.MQConst;
-import com.liquidnet.service.consumer.kylin.service.IBaseDao;
+import com.liquidnet.service.goblin.dto.GoblinQueueBizMongoDto;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.mongodb.core.MongoTemplate;
@@ -23,10 +23,12 @@ import java.util.HashMap;
  */
 @Slf4j
 public abstract class AbstractMdbRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
+    @Autowired
+    public StringRedisTemplate stringRedisTemplate;
     @Autowired
     private MongoTemplate mongoTemplate;
     @Autowired
-    public StringRedisTemplate stringRedisTemplate;
+    private RedisUtil redisUtil;
 
     @Override
     public void onMessage(MapRecord<String, String, String> message) {
@@ -50,11 +52,22 @@ public abstract class AbstractMdbRedisReceiver implements StreamListener<String,
     private boolean consumerMessageHandler(String msg) {
         boolean aBoolean = false;
         try {
-            SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
-            if (sqlMessage == null) {
+            GoblinQueueBizMongoDto queueBizMongoDto = JsonUtils.fromJson(msg, GoblinQueueBizMongoDto.class);
+            if (null == queueBizMongoDto) {
                 aBoolean = true;
             } else {
-//                mongoTemplate
+                String bizId = queueBizMongoDto.getBizId();
+                int opType = queueBizMongoDto.getOpType();
+
+                Object o = redisUtil.get(bizId);
+                switch (opType) {
+                    case 1:// insert
+                        mongoTemplate.insert("");
+                        break;
+                    case 2:// update
+                        mongoTemplate.getCollection("");
+                        break;
+                }
             }
         } catch (Exception e) {
             log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);

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

@@ -453,7 +453,7 @@ public class GoblinNftOrderServiceImpl implements IGoblinNftOrderService {
             // nftOrderUtils.addNftOrderList(uid, orderVo.getOrderId());
 
             // mongo
-            goblinOrderUtils.setMongoList(nftOrder.getOrderId(), 1, 1, 1);
+            goblinOrderUtils.setMongoList(nftOrder.getOrderId(), 1, 1);
 //            goblinMongoUtils.setGoblinNftOrderVo(orderVo);
 
             // 执行sql
@@ -646,7 +646,7 @@ public class GoblinNftOrderServiceImpl implements IGoblinNftOrderService {
             // redis
             nftOrderUtils.setNftOrder(orderVo);
             // mongo
-            goblinOrderUtils.setMongoList(orderId, 1, 1, 1);
+            goblinOrderUtils.setMongoList(orderId, 1, 1);
 //            goblinMongoUtils.updateGoblinNftOrderVo(orderVo);
             // mysql
             LinkedList<String> sqls = CollectionUtil.linkedListString();

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

@@ -226,14 +226,13 @@ public class GoblinOrderUtils {
      * @param bizId 业务id 例如orderId
      * @param bizType 业务类型 1NFT订单 2...
      * @param opType 操作类型 1创建 2修改
-     * @param listType 队列类型 1即时性要求高队列 2即时性要求不高队列
      */
-    public void setMongoList(String bizId, int bizType, int opType, int listType) {
+    public void setMongoList(String bizId, int bizType, int opType) {
         GoblinQueueBizMongoDto goblinQueueBizMongoDto = GoblinQueueBizMongoDto.getNew();
             goblinQueueBizMongoDto.setBizId(bizId);
-            goblinQueueBizMongoDto.setBizType(bizType);
+            goblinQueueBizMongoDto.setBizType(bizType + "");
             goblinQueueBizMongoDto.setOpType(opType);
-            goblinQueueBizMongoDto.setListType(listType);
+//            goblinQueueBizMongoDto.setListType(listType);
             queueUtils.sendMsgByRedis(MQConst.GoblinQueue.BIZ_NFT_MONGO.getKey(), goblinQueueBizMongoDto.toJson());
     }