opt;

liquidnet-bus-common/liquidnet-common-cache/liquidnet-common-cache-redis/src/main/java/com.liquidnet.common.cache/redis/config/RedisStreamConfig.java

@@ -33,6 +33,7 @@ public class RedisStreamConfig {
         try {
             if (stringRedisTemplate.hasKey(key)) {
                 log.info("redis stream exist[{},{}]", key, group);
+                stringRedisTemplate.opsForStream().createGroup(key, group);
             } else {
                 log.info("redis stream init[{},{}]", key, group);
                 HashMap<String, String> map = new HashMap<>();
@@ -43,7 +44,7 @@ public class RedisStreamConfig {
                 stringObjectObjectStreamOperations.delete(key, recordId.getValue());
             }
         } catch (Exception e) {
-            log.error("Ex:redis stream init [{},{}]", key, group, e);
+            log.error("Ex:redis stream init [{},{}],ex:{}", key, group, e.getMessage());
         }
     }
 

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

@@ -2,8 +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 com.liquidnet.service.consumer.kylin.receiver.*;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -24,10 +23,64 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
     @Autowired
     StringRedisTemplate stringRedisTemplate;
     @Autowired
+    ConsumerGoblinSqlStoreRdsReceiver consumerGoblinSqlStoreRdsReceiver;
+    @Autowired
+    ConsumerGoblinSqlGoodsRdsReceiver consumerGoblinSqlGoodsRdsReceiver;
+    @Autowired
+    ConsumerGoblinSqlCouponRdsReceiver consumerGoblinSqlCouponRdsReceiver;
+    @Autowired
     ConsumerGoblinSqlArtworkGenRdsReceiver consumerGoblinSqlArtworkGenRdsReceiver;
     @Autowired
     ConsumerGoblinNftOrderRdsReceiver consumerGoblinNftOrderRdsReceiver;
 
+    @Bean// 店铺相关
+    public List<Subscription> subscriptionSqlStore(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_STORE;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 2; 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()), consumerGoblinSqlStoreRdsReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+
+    @Bean// 商品相关
+    public List<Subscription> subscriptionSqlGoods(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_GOODS;
+        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()), consumerGoblinSqlGoodsRdsReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+
+    @Bean// 店铺优惠券相关
+    public List<Subscription> subscriptionSqlStoreCoupon(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_COUPON;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 10; 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()), consumerGoblinSqlCouponRdsReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+
     @Bean// 藏品生成
     public List<Subscription> subscriptionSqlArtworkGen(RedisConnectionFactory factory) {
         List<Subscription> subscriptionList = new ArrayList<>();
@@ -45,7 +98,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
     }
 
     @Bean// NFT订单
-    public List<Subscription> subscriptionBizNftOrder(RedisConnectionFactory factory) {
+    public List<Subscription> subscriptionSqlNftOrder(RedisConnectionFactory factory) {
         List<Subscription> subscriptionList = new ArrayList<>();
         MQConst.GoblinQueue stream = MQConst.GoblinQueue.GOBLIN_NFT_ORDER;
         this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());