队列更改

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerCommonSqlRedisStreamConfig.java

@@ -2,6 +2,7 @@ package com.liquidnet.service.consumer.kylin.config;
 
 import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
 import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.kylin.receiver.ConsumerGoblinNftOrderRdsReceiver;
 import com.liquidnet.service.consumer.kylin.receiver.ConsumerGoblinSqlArtworkGenRdsReceiver;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -24,6 +25,8 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
     StringRedisTemplate stringRedisTemplate;
     @Autowired
     ConsumerGoblinSqlArtworkGenRdsReceiver consumerGoblinSqlArtworkGenRdsReceiver;
+    @Autowired
+    ConsumerGoblinNftOrderRdsReceiver consumerGoblinNftOrderRdsReceiver;
 
     @Bean// 藏品生成
     public List<Subscription> subscriptionSqlArtworkGen(RedisConnectionFactory factory) {
@@ -40,4 +43,20 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
         }
         return subscriptionList;
     }
+
+    @Bean// NFT订单
+    public List<Subscription> subscriptionBizNftOrder(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GoblinQueue stream = MQConst.GoblinQueue.GOBLIN_NFT_ORDER;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 3; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinNftOrderRdsReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
 }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerGoblinOrderCPRedisStreamConfig.java

@@ -2,7 +2,6 @@ package com.liquidnet.service.consumer.kylin.config;
 
 import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
 import com.liquidnet.service.base.constant.MQConst;
-import com.liquidnet.service.consumer.kylin.receiver.ConsumerGoblinNftOrderRdsReceiver;
 import com.liquidnet.service.consumer.kylin.receiver.ConsumerGoblinOrderCPRdsReceiver;
 import lombok.var;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -21,9 +20,6 @@ public class ConsumerGoblinOrderCPRedisStreamConfig extends RedisStreamConfig {
     @Autowired
     ConsumerGoblinOrderCPRdsReceiver consumerGoblinOrderCPRdsReceiver;
 
-    @Autowired
-    ConsumerGoblinNftOrderRdsReceiver consumerGoblinNftOrderRdsReceiver;
-
     /**
      * 商城订单
      */
@@ -32,14 +28,6 @@ public class ConsumerGoblinOrderCPRedisStreamConfig extends RedisStreamConfig {
                 StreamOffset.create(MQConst.GoblinQueue.GOBLIN_ORDER_CREATE_PAY.getKey(), ReadOffset.lastConsumed()), consumerGoblinOrderCPRdsReceiver);
     }
 
-    /**
-     * NFT订单
-     */
-    private Subscription receiveGoblinNftOrder(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
-        return listenerContainer.receiveAutoAck(Consumer.from(MQConst.GoblinQueue.GOBLIN_NFT_ORDER.getGroup(), getConsumerName(MQConst.GoblinQueue.GOBLIN_NFT_ORDER.name() + t)),
-                StreamOffset.create(MQConst.GoblinQueue.GOBLIN_NFT_ORDER.getKey(), ReadOffset.lastConsumed()), consumerGoblinNftOrderRdsReceiver);
-    }
-
     /* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
 
     /*-------------------------------------- 商城订单 --------------------------------------*/
@@ -68,31 +56,5 @@ public class ConsumerGoblinOrderCPRedisStreamConfig extends RedisStreamConfig {
         return subscription;
     }
 
-    /*-------------------------------------- NFT订单 --------------------------------------*/
-
-    @Bean
-    public Subscription subscriptionGoblinNftOrder0(RedisConnectionFactory factory) {
-        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
-        var subscription = receiveGoblinNftOrder(listenerContainer, 0);
-        listenerContainer.start();
-        return subscription;
-    }
-
-    @Bean
-    public Subscription subscriptionGoblinNftOrder1(RedisConnectionFactory factory) {
-        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
-        var subscription = receiveGoblinNftOrder(listenerContainer, 1);
-        listenerContainer.start();
-        return subscription;
-    }
-
-    @Bean
-    public Subscription subscriptionGoblinNftOrder2(RedisConnectionFactory factory) {
-        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
-        var subscription = receiveGoblinNftOrder(listenerContainer, 2);
-        listenerContainer.start();
-        return subscription;
-    }
-
     /* -------------------------------------------------------- ｜  */
 }