redis-队列多数据源

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

@@ -39,6 +39,8 @@ public abstract class AbstractRedisConfig {
 
     public Map<Integer, RedisTemplate<String, Object>> redisTemplateMap = new HashMap<>();
 
+    private StringRedisTemplate stringRedisTemplate = new StringRedisTemplate();
+
     abstract String getHost();
     abstract int getPort();
     abstract String getPassword();
@@ -69,6 +71,21 @@ public abstract class AbstractRedisConfig {
             redisTemplateMap.put(defaultDb, getRedisTemplate(defaultDb));
         }
         log.info("###### END 初始化 Redis 连接池 END ######");
+
+        log.info("###### START 初始化 Redis-queue "+this.getClass().getSimpleName()+"连接池 START ######");
+        //初始化队列
+        LettuceConnectionFactory factory = null;
+        if(totalDbs==1){
+            log.info("###### 正在加载Redis-queue-db-" + defaultDb + " ######");
+            factory = getDbFactory(defaultDb);
+        }else{
+            log.info("###### 正在加载Redis-queue-db-" + (totalDbs-1) + " ######");
+            factory = getDbFactory(totalDbs-1);
+        }
+
+        stringRedisTemplate.setConnectionFactory(factory);
+        stringRedisTemplate.afterPropertiesSet();
+        log.info("###### END 初始化 Redis-queue 连接池 END ######");
     }
 
     private RedisTemplate<String, Object> getRedisTemplate(int dbNo) {
@@ -127,22 +144,26 @@ public abstract class AbstractRedisConfig {
 //    }
 
     public StringRedisTemplate getStringRedisTemplate(){
-        if(StringUtils.isEmpty(getHost())||getHost().equalsIgnoreCase("null")){
-            log.info("无配置，不需要初始化！");
-            return null;
-        }
-        LettuceConnectionFactory factory = null;
-        if(totalDbs==1){
-            factory = getDbFactory(defaultDb);
-        }else{
-            factory = getDbFactory(totalDbs-1);
-        }
-        StringRedisTemplate redisTemplate = new StringRedisTemplate();
-        redisTemplate.setConnectionFactory(factory);
-        redisTemplate.afterPropertiesSet();
-        return redisTemplate;
+        return stringRedisTemplate;
     }
 
+//    public StringRedisTemplate getDefaultStringRedisTemplate(){
+//        if(StringUtils.isEmpty(getHost())||getHost().equalsIgnoreCase("null")){
+//            log.info("无配置，不需要初始化！");
+//            return null;
+//        }
+//        LettuceConnectionFactory factory = null;
+//        if(totalDbs==1){
+//            factory = getDbFactory(defaultDb);
+//        }else{
+//            factory = getDbFactory(totalDbs-1);
+//        }
+//        StringRedisTemplate stringRedisTemplate = new StringRedisTemplate();
+//        stringRedisTemplate.setConnectionFactory(factory);
+//        stringRedisTemplate.afterPropertiesSet();
+//        return stringRedisTemplate;
+//    }
+
     private void setSerializer(RedisTemplate<String, Object> template) {
         Jackson2JsonRedisSerializer j2jrs = new Jackson2JsonRedisSerializer(Object.class);
         ObjectMapper om = new ObjectMapper();

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

+package com.liquidnet.common.cache.redis.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.List;
+
+/**
+ * @author AnJiabin <anjiabin@zhengzai.tv>
+ * @version V1.0
+ * @Description: 队列redis
+ * @class: RedisStreamQueueConfig
+ * @Package com.liquidnet.common.cache.redis.config
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/6/14 15:10
+ */
+@Slf4j
+@Configuration
+public class RedisQueueConfig extends AbstractRedisConfig{
+    @Value("${spring.redis.queue.host:null}")
+    private String host;
+    @Value("${spring.redis.queue.port:-1}")
+    private int port;
+    @Value("${spring.redis.queue.password:null}")
+    private String password;
+    @Value("${spring.redis.queue.lettuce.pool.max-active:-1}")
+    private int maxActive;
+    @Value("${spring.redis.queue.lettuce.pool.max-idle:-1}")
+    private int maxIdle;
+    @Value("${spring.redis.queue.lettuce.pool.min-idle:-1}")
+    private int minIdle;
+    @Value("${spring.redis.queue.lettuce.pool.max-wait:-1}")
+    private int maxWait;
+
+    @Value("${spring.redis.queue.dbs:${spring.redis.queue.database:-1}}")
+    private List<Integer> dbs;
+
+    String getHost(){
+        return this.host;
+    }
+
+    int getPort(){
+        return this.port;
+    }
+    String getPassword(){
+        return this.password;
+    }
+    int getMaxActive(){
+        return this.maxActive;
+    }
+    int getMaxIdle(){
+        return this.maxIdle;
+    }
+    int getMinIdle(){
+        return this.minIdle;
+    }
+    int getMaxWait(){
+        return this.maxWait;
+    }
+    List<Integer> getDbs(){
+        return this.dbs;
+    }
+}
+

liquidnet-bus-common/liquidnet-common-cache/liquidnet-common-cache-redis/src/main/java/com.liquidnet.common.cache/redis/util/RedisDataSourceUtil.java

@@ -35,6 +35,9 @@ public class RedisDataSourceUtil {
     @Autowired
     private RedisGalaxyUtil redisGalaxyUtil;
 
+    @Autowired
+    private RedisQueueUtil redisQueueUtil;
+
     public AbstractRedisUtil getRedisKylinUtil() {
         return this.redisKylinUtil;
     }
@@ -66,4 +69,8 @@ public class RedisDataSourceUtil {
     public AbstractRedisUtil getRedisGalaxyUtil(){
         return this.redisGalaxyUtil;
     }
+
+    public AbstractRedisUtil getRedisQueueUtil(){
+        return this.redisQueueUtil;
+    }
 }

liquidnet-bus-common/liquidnet-common-cache/liquidnet-common-cache-redis/src/main/java/com.liquidnet.common.cache/redis/util/RedisQueueUtil.java

+package com.liquidnet.common.cache.redis.util;
+
+import com.liquidnet.common.cache.redis.config.AbstractRedisConfig;
+import com.liquidnet.common.cache.redis.config.RedisQueueConfig;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author AnJiabin <anjiabin@zhengzai.tv>
+ * @version V1.0
+ * @Description: TODO
+ * @class: RedisQueueUtil
+ * @Package com.liquidnet.common.cache.redis.util
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/6/14 15:30
+ */
+@Slf4j
+@Component("redisQueueUtil")
+public final class RedisQueueUtil extends AbstractRedisUtil{
+    @Autowired
+    private RedisQueueConfig redisConfig;
+
+    @Override
+    public int getDbs() {
+        log.info("RedisQueueUtil.totalDbs===",redisConfig.totalDbs);
+        return redisConfig.totalDbs;
+    }
+
+    @Override
+    AbstractRedisConfig getRedisConfig() {
+        return this.redisConfig;
+    }
+}
\ No newline at end of file

liquidnet-bus-config/liquidnet-config/application-dev.yml

@@ -31,6 +31,11 @@ liquidnet:
   knife4j:
     disable: false
   redis:
+    queue:
+      database: 15
+      host: 39.107.71.112
+      port: 6379
+      password: 3Xa%8p
     kylin:
       dbs: 0,16
       database: 15

liquidnet-bus-config/liquidnet-config/application-test.yml

@@ -31,6 +31,11 @@ liquidnet:
   knife4j:
     disable: false
   redis:
+    queue:
+      database: 255
+      host: r-2zeucai3yj2t0f4nmzpd.redis.rds.aliyuncs.com
+      port: 6380
+      password: 7eoK2XehKqF1
     kylin:
       database: 255
       dbs: 0,256

liquidnet-bus-config/liquidnet-config/liquidnet-service-goblin.yml

@@ -75,6 +75,17 @@ spring:
       - org.springframework.cloud.bus.BusAutoConfiguration
       - org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration
   redis:
+    queue:
+      database: ${liquidnet.redis.queue.database}
+      port: ${liquidnet.redis.queue.port}
+      host: ${liquidnet.redis.queue.host}
+      password: ${liquidnet.redis.queue.password}
+      lettuce:
+        pool:
+          max-active: 8
+          max-wait: -1
+          max-idle: 8
+          min-idle: 0
     database: 15
     dbs: ${liquidnet.redis.goblin.dbs}
     port: ${liquidnet.redis.goblin.port}

liquidnet-bus-service/liquidnet-service-goblin/liquidnet-service-goblin-common/src/main/java/com/liquidnet/service/galaxy/utils/QueueUtil.java

 package com.liquidnet.service.galaxy.utils;
 
+import com.liquidnet.common.cache.redis.util.RedisDataSourceUtil;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
@@ -16,10 +17,10 @@ import org.springframework.stereotype.Component;
 @Component
 public class QueueUtil extends AbstractQueueUtil{
     @Autowired
-    StringRedisTemplate stringRedisTemplate;
+    private RedisDataSourceUtil redisDataSourceUtil;
 
     @Override
     StringRedisTemplate getStringRedisTemplate() {
-        return stringRedisTemplate;
+        return redisDataSourceUtil.getRedisQueueUtil().getStringRedisTemplate();
     }
 }

liquidnet-bus-service/liquidnet-service-goblin/liquidnet-service-goblin-impl/src/main/java/com/liquidnet/service/goblin/service/impl/inner/GoblinQueBizArtworkClqService.java

 package com.liquidnet.service.goblin.service.impl.inner;
 
+import com.liquidnet.common.cache.redis.util.RedisDataSourceUtil;
 import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.commons.lang.util.JsonUtils;
 import com.liquidnet.service.base.ResponseDto;
@@ -21,7 +22,6 @@ import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.data.redis.connection.stream.StreamRecords;
-import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
@@ -40,8 +40,9 @@ public class GoblinQueBizArtworkClqService {
     GoblinMongoUtils goblinMongoUtils;
     @Autowired
     MongoTemplate mongoTemplate;
+
     @Autowired
-    StringRedisTemplate stringRedisTemplate;
+    private RedisDataSourceUtil redisDataSourceUtil;
 
     @Resource(name = "galaxyArtworkServiceImpl")
     private IGalaxyArtworkService galaxyArtworkService;
@@ -90,7 +91,7 @@ public class GoblinQueBizArtworkClqService {
             updateGoodsSkuNftObjs.add(new Object[]{1, seriesId, txHash, nftHash, chainTimestamp, now, skuId});
 
             sqlUpdateMap.put(MQConst.QUEUE_MESSAGE_KEY, SqlMapping.gets(toMqSqls, updateGoodsSkuNftObjs));
-            stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(sqlUpdateMap).withStreamKey(MQConst.GoblinQueue.SQL_GOODS.getKey()));
+            redisDataSourceUtil.getRedisQueueUtil().getStringRedisTemplate().opsForStream().add(StreamRecords.mapBacked(sqlUpdateMap).withStreamKey(MQConst.GoblinQueue.SQL_GOODS.getKey()));
         } catch (Exception e) {
             log.error("Ex.NFT声明查询：处理异常[skuId={}]", skuId, e);
             queueUtils.sendMsgByRedis(MQConst.GoblinQueue.BIZ_ARTWORK_CLQ.getKey(), skuId);

liquidnet-bus-service/liquidnet-service-goblin/liquidnet-service-goblin-impl/src/main/java/com/liquidnet/service/goblin/service/impl/inner/GoblinQueBizArtworkUplService.java

 package com.liquidnet.service.goblin.service.impl.inner;
 
+import com.liquidnet.common.cache.redis.util.RedisDataSourceUtil;
 import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.commons.lang.util.JsonUtils;
 import com.liquidnet.service.base.ResponseDto;
@@ -25,7 +26,6 @@ import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.data.redis.connection.stream.StreamRecords;
 import org.springframework.data.redis.core.StreamOperations;
-import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
@@ -44,8 +44,9 @@ public class GoblinQueBizArtworkUplService {
     GoblinMongoUtils goblinMongoUtils;
     @Autowired
     MongoTemplate mongoTemplate;
+
     @Autowired
-    StringRedisTemplate stringRedisTemplate;
+    private RedisDataSourceUtil redisDataSourceUtil;
 
     @Resource(name = "galaxyArtworkServiceImpl")
     private IGalaxyArtworkService galaxyArtworkService;
@@ -118,7 +119,7 @@ public class GoblinQueBizArtworkUplService {
 
             sqlUpdateMap.put(MQConst.QUEUE_MESSAGE_KEY, SqlMapping.gets(toMqSqls, updateGoodsSkuNftObjs));
 
-            StreamOperations<String, Object, Object> streamOperations = stringRedisTemplate.opsForStream();
+            StreamOperations<String, Object, Object> streamOperations = redisDataSourceUtil.getRedisQueueUtil().getStringRedisTemplate().opsForStream();
             streamOperations.add(StreamRecords.mapBacked(sqlUpdateMap).withStreamKey(MQConst.GoblinQueue.SQL_GOODS.getKey()));
 
             HashMap<String, String> toQueueBeClaimQueryMsg = CollectionUtil.mapStringString();

liquidnet-bus-service/liquidnet-service-goblin/liquidnet-service-goblin-impl/src/main/java/com/liquidnet/service/goblin/util/QueueUtils.java

 package com.liquidnet.service.goblin.util;
 
+import com.liquidnet.common.cache.redis.util.RedisDataSourceUtil;
 import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.service.base.constant.MQConst;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.connection.stream.StreamRecords;
-import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import java.util.HashMap;
@@ -14,8 +14,9 @@ public class QueueUtils {
 
     //    @Autowired
 //    private RabbitTemplate rabbitTemplate;
+
     @Autowired
-    StringRedisTemplate stringRedisTemplate;
+    private RedisDataSourceUtil redisDataSourceUtil;
 
     /**
      * 发送消息 - RABBIT
@@ -37,7 +38,7 @@ public class QueueUtils {
     public void sendMsgByRedis(String streamKey, String jsonMsg) {
         HashMap<String, String> map = CollectionUtil.mapStringString();
         map.put(MQConst.QUEUE_MESSAGE_KEY, jsonMsg);
-        stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(streamKey));
+        redisDataSourceUtil.getRedisQueueUtil().getStringRedisTemplate().opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(streamKey));
     }
 
     /**
@@ -55,7 +56,7 @@ public class QueueUtils {
             map.put(MQConst.QUEUE_MESSAGE_KEY, xlsPath);
             map.put("type", type);
             map.put("skuId", skuId);
-            stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey()));
+            redisDataSourceUtil.getRedisQueueUtil().getStringRedisTemplate().opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey()));
         }
     }
 
@@ -74,6 +75,6 @@ public class QueueUtils {
         map.put("listId", listId);
         map.put("type", type);
         map.put("skuId", skuId);
-        stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey()));
+        redisDataSourceUtil.getRedisQueueUtil().getStringRedisTemplate().opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(MQConst.GoblinQueue.GOBLIN_XLS_OPERA.getKey()));
     }
 }

liquidnet-bus-service/liquidnet-service-goblin/liquidnet-service-goblin-impl/src/test/java/com/liquidnet/service/goblin/test/TestResisQueue.java

+package com.liquidnet.service.goblin.test;
+
+import com.liquidnet.common.cache.redis.util.RedisDataSourceUtil;
+import com.liquidnet.commons.lang.util.CollectionUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.data.redis.connection.stream.StreamRecords;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import java.util.HashMap;
+
+/**
+ * @author AnJiabin <anjiabin@zhengzai.tv>
+ * @version V1.0
+ * @Description: TODO
+ * @class: TestResisQueue
+ * @Package com.liquidnet.service.goblin.test
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/6/14 15:42
+ */
+@Slf4j
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest
+public class TestResisQueue {
+    @Autowired
+    private RedisDataSourceUtil redisDataSourceUtil;
+
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    /**
+     * 兼容测试
+     */
+    @Test
+    public void oldPringRedisInitInfo(){
+        redisDataSourceUtil.getRedisKylinUtil().set("testNldRedis","12345");
+        redisDataSourceUtil.getRedisKylinUtil().set("testoldRedis","12345");
+        System.out.println("getRedisKylinUtil.get==="+ redisDataSourceUtil.getRedisKylinUtil().get("testoldRedis"));
+        System.out.println("getRedisDragonUtil.totalDbs==="+ redisDataSourceUtil.getRedisDragonUtil().getDbs());
+    }
+
+    /**
+     * 兼容测试队列
+     */
+    @Test
+    public void oldGetRedisQueue(){
+        String json = "{\n" +
+                "  \"nftOrderPayId\": \"1234567890-order-022-008\",\n" +
+                "  \"routerType\": \"zxinchain\",\n" +
+                "  \"skuId\": \"1234567890-sku-022\",\n" +
+                "  \"userId\": \"123456789-user-002\",\n" +
+                "  \"buyTimestamp\":\"2022-04-12 13:50:12\"\n" +
+                "}";
+        HashMap<String, String> map = CollectionUtil.mapStringString();
+        map.put("message", json);
+        log.info("stringRedisTemplate.stream===",stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey("goblintest_old:stream:dragon-pay")));
+
+        json = "{\n" +
+                "  \"nftOrderPayId\": \"1234567890-queue-022-008\",\n" +
+                "  \"routerType\": \"zxinchain\",\n" +
+                "  \"skuId\": \"1234567890-sku-022\",\n" +
+                "  \"userId\": \"123456789-user-002\",\n" +
+                "  \"buyTimestamp\":\"2022-04-12 13:50:12\"\n" +
+                "}";
+        map.put("message", json);
+        log.info("getRedisQueueUtil.stream===",redisDataSourceUtil.getRedisQueueUtil().getStringRedisTemplate().opsForStream().add(StreamRecords.mapBacked(map).withStreamKey("goblintest_new:stream:dragon-pay")));
+
+//        log.info("getRedisKylinUtil.stream===",redisDataSourceUtil.getRedisKylinUtil().getStringRedisTemplate().opsForStream().size("dragon:stream:dragon-pay"));
+//        log.info("getRedisDragonUtil.stream===",redisDataSourceUtil.getRedisDragonUtil().getStringRedisTemplate().opsForStream().size("dragon:stream:dragon-pay"));
+//        log.info("getRedisSweetUtil.stream===",redisDataSourceUtil.getRedisSweetUtil().getStringRedisTemplate().opsForStream().size("abc"));
+//        log.info("getRedisAdamUtil.stream===",redisDataSourceUtil.getRedisAdamUtil().getStringRedisTemplate().opsForStream().size("abc"));
+//        log.info("getRedisCandyUtil.stream===",redisDataSourceUtil.getRedisCandyUtil().getStringRedisTemplate().opsForStream().size("abc"));
+    }
+}