重构redis维表cache=all的结构

dapeng · dapeng · commit da409ef8f086 · 2020-03-11T16:08:25.000+08:00
diff --git a/docs/redisSide.md b/docs/redisSide.md
@@ -4,6 +4,7 @@
  CREATE TABLE tableName(
      colName cloType,
      ...
+     PRIMARY KEY(colName1,colName2) ,
      PERIOD FOR SYSTEM_TIME
   )WITH(
      type ='redis',
@@ -27,9 +28,10 @@
 | tableName | 注册到flink的表名称(可选填;不填默认和redis对应的表名称相同)|
 | colName | 列名称，维表列名格式 表名:主键名:主键值:列名]|
 | colType | 列类型，当前只支持varchar|
+| PRIMARY KEY |主键，多个字段做为联合主键时以逗号分隔
 | PERIOD FOR SYSTEM_TIME | 关键字表明该定义的表为维表信息|
   
-## 3.参数
+## 4.参数
 
 |参数名称|含义|是否必填|默认值|
 |----|---|---|----|
@@ -51,7 +53,7 @@
     * cacheTTLMs:缓存的过期时间(ms)
   * ALL: 缓存全量表数据
 
-## 4.样例
+## 5.样例
 ```
 create table sideTable(
     channel varchar,
@@ -70,5 +72,10 @@ create table sideTable(
  );
 
 ```
+## 6.缓存redis的存储结构规则
+```
+redis使用散列类型 hash 数据结构，key=tableName_primaryKey1_primaryKey2,value={column1=value1, column2=value2}
+如果以班级class表为例，id和name作为联合主键，那么redis的结构为 <class_1_john ,{id=1, name=john, age=12}>
+```
 
 
diff --git a/redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java b/redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java
@@ -19,10 +19,14 @@
 package com.dtstack.flink.sql.side.redis;
 
 import com.dtstack.flink.sql.side.*;
+import com.dtstack.flink.sql.side.redis.enums.RedisType;
 import com.dtstack.flink.sql.side.redis.table.RedisSideReqRow;
 import com.dtstack.flink.sql.side.redis.table.RedisSideTableInfo;
 import com.esotericsoftware.minlog.Log;
 import org.apache.calcite.sql.JoinType;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang.StringUtils;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.Maps;
@@ -103,104 +107,51 @@ public void flatMap(Row row, Collector<Row> out) throws Exception {
             String columnName = sideInfo.getEqualFieldList().get(conValIndex);
             inputParams.put(columnName, equalObj.toString());
         }
-        String key = buildKey(inputParams);
+        String key = buildCacheKey(inputParams);
+        if(StringUtils.isBlank(key)){
+            return;
+        }
 
         Map<String, String> cacheMap = cacheRef.get().get(key);
-
-        if (cacheMap == null){
-            if(sideInfo.getJoinType() == JoinType.LEFT){
-                Row data = fillData(row, null);
-                out.collect(data);
-            }else{
+        if(MapUtils.isEmpty(cacheMap)){
+            if(sideInfo.getJoinType() != JoinType.LEFT){
                 return;
             }
-
+            Row data = fillData(row, null);
+            out.collect(data);
             return;
-        }
 
+        }
         Row newRow = fillData(row, cacheMap);
         out.collect(newRow);
     }
 
-    private String buildKey(Map<String, String> inputParams) {
-        String tableName = tableInfo.getTableName();
-        StringBuilder key = new StringBuilder();
-        for (int i=0; i<inputParams.size(); i++){
-            key.append(tableName).append(":").append(inputParams.keySet().toArray()[i]).append(":")
-                    .append(inputParams.get(inputParams.keySet().toArray()[i]));
+    private String buildCacheKey(Map<String, String> refData) {
+        StringBuilder keyBuilder = new StringBuilder(tableInfo.getTableName());
+        List<String> primaryKeys = tableInfo.getPrimaryKeys();
+        for(String primaryKey : primaryKeys){
+            if(!refData.containsKey(primaryKey)){
+                return null;
+            }
+            keyBuilder.append("_").append(refData.get(primaryKey));
         }
-        return key.toString();
+        return keyBuilder.toString();
     }
 
+
     private void loadData(Map<String, Map<String, String>> tmpCache) throws SQLException {
         JedisCommands jedis = null;
-
         try {
-            for(int i=0; i<CONN_RETRY_NUM; i++){
-
-                try{
-                    jedis = getJedis(tableInfo);
-                    break;
-                }catch (Exception e){
-                    if(i == CONN_RETRY_NUM - 1){
-                        throw new RuntimeException("", e);
-                    }
-
-                    try {
-                        String jedisInfo = "url:" + tableInfo.getUrl() + ",pwd:" + tableInfo.getPassword() + ",database:" + tableInfo.getDatabase();
-                        LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + jedisInfo);
-                        Thread.sleep(5 * 1000);
-                    } catch (InterruptedException e1) {
-                        LOG.error("", e1);
-                    }
-                }
-            }
-
-            if (tableInfo.getRedisType() != 3){
-                String perKey = tableInfo.getTableName() + "*";
-                Set<String> keys = ((Jedis) jedis).keys(perKey);
-                List<String> newPerKeys = new LinkedList<>();
-                for (String key : keys){
-                    String[] splitKey = key.split(":");
-                    String newKey = splitKey[0] + ":" + splitKey[1] + ":" + splitKey[2];
-                    newPerKeys.add(newKey);
-                }
-                List<String> list = newPerKeys.stream().distinct().collect(Collectors.toList());
-                for(String key : list){
-                    Map<String, String> kv = Maps.newHashMap();
-                    String[] primaryKv = key.split(":");
-                    kv.put(primaryKv[1], primaryKv[2]);
-                    String pattern = key + "*";
-                    Set<String> realKeys = ((Jedis) jedis).keys(pattern);
-                    for (String realKey : realKeys){
-                        kv.put(realKey.split(":")[3], jedis.get(realKey));
-                    }
-                    tmpCache.put(key, kv);
-                }
-            } else {
-                String perKey = tableInfo.getTableName() + "*";
-                Set<String> keys = keys((JedisCluster) jedis, perKey);
-                List<String> newPerKeys = new LinkedList<>();
-                for (String key : keys){
-                    String[] splitKey = key.split(":");
-                    String newKey = splitKey[0] + ":" + splitKey[1] + ":" + splitKey[2];
-                    newPerKeys.add(newKey);
-                }
-                List<String> list = newPerKeys.stream().distinct().collect(Collectors.toList());
-                for(String key : list){
-                    Map<String, String> kv = Maps.newHashMap();
-                    String[] primaryKv = key.split(":");
-                    kv.put(primaryKv[1], primaryKv[2]);
-                    String pattern = key + "*";
-                    Set<String> realKeys = keys((JedisCluster) jedis, pattern);
-                    for (String realKey : realKeys){
-                        kv.put(realKey.split(":")[3], jedis.get(realKey));
-                    }
-                    tmpCache.put(key, kv);
-                }
+            String keyPattern = tableInfo.getTableName() + "*";
+            jedis = getJedisWithRetry(CONN_RETRY_NUM);
+            Set<String> keys = getRedisKeys(RedisType.parse(tableInfo.getRedisType()), jedis, keyPattern);
+            if(CollectionUtils.isEmpty(keys)){
+                return;
             }
-
-
+            JedisCommands redis = jedis;
+            keys.forEach(k ->{
+                tmpCache.put(k, redis.hgetAll(k));
+            });
         } catch (Exception e){
             LOG.error("", e);
         } finally {
@@ -245,33 +196,58 @@ private JedisCommands getJedis(RedisSideTableInfo tableInfo) {
         }
         JedisCommands jedis = null;
         GenericObjectPoolConfig poolConfig = setPoolConfig(tableInfo.getMaxTotal(), tableInfo.getMaxIdle(), tableInfo.getMinIdle());
-        switch (tableInfo.getRedisType()){
+        switch (RedisType.parse(tableInfo.getRedisType())){
             //单机
-            case 1:
+            case STANDALONE:
                 pool = new JedisPool(poolConfig, firstIp, Integer.parseInt(firstPort), timeout, password, Integer.parseInt(database));
                 jedis = pool.getResource();
                 break;
             //哨兵
-            case 2:
+            case SENTINEL:
                 jedisSentinelPool = new JedisSentinelPool(tableInfo.getMasterName(), ipPorts, poolConfig, timeout, password, Integer.parseInt(database));
                 jedis = jedisSentinelPool.getResource();
                 break;
             //集群
-            case 3:
+            case CLUSTER:
                 jedis = new JedisCluster(addresses, timeout, timeout,1, poolConfig);
+            default:
+                break;
         }
 
         return jedis;
     }
 
-    private Set<String> keys(JedisCluster jedisCluster, String pattern){
+    private JedisCommands getJedisWithRetry(int retryNum) {
+        while (retryNum-- > 0){
+            try {
+                return getJedis(tableInfo);
+            } catch (Exception e) {
+                if(retryNum <= 0){
+                    throw new RuntimeException("getJedisWithRetry error", e);
+                }
+                try {
+                    String jedisInfo = "url:" + tableInfo.getUrl() + ",pwd:" + tableInfo.getPassword() + ",database:" + tableInfo.getDatabase();
+                    LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + jedisInfo);
+                    Thread.sleep(5 * 1000);
+                } catch (InterruptedException e1) {
+                    LOG.error("", e1);
+                }
+            }
+        }
+        return null;
+    }
+
+    private Set<String> getRedisKeys(RedisType redisType, JedisCommands jedis, String keyPattern){
+        if(!redisType.equals(RedisType.CLUSTER)){
+            return ((Jedis) jedis).keys(keyPattern);
+        }
         Set<String> keys = new TreeSet<>();
-        Map<String, JedisPool> clusterNodes = jedisCluster.getClusterNodes();
+        Map<String, JedisPool> clusterNodes = ((JedisCluster)jedis).getClusterNodes();
         for(String k : clusterNodes.keySet()){
             JedisPool jp = clusterNodes.get(k);
             Jedis connection = jp.getResource();
             try {
-                keys.addAll(connection.keys(pattern));
+                keys.addAll(connection.keys(keyPattern));
             } catch (Exception e){
                 LOG.error("Getting keys error: {}", e);
             } finally {
diff --git a/redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java b/redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java
@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.enums.ECacheContentType;
 import com.dtstack.flink.sql.side.*;
 import com.dtstack.flink.sql.side.cache.CacheObj;
+import com.dtstack.flink.sql.side.redis.enums.RedisType;
 import com.dtstack.flink.sql.side.redis.table.RedisSideReqRow;
 import com.dtstack.flink.sql.side.redis.table.RedisSideTableInfo;
 import io.lettuce.core.RedisClient;
@@ -86,28 +87,30 @@ private void buildRedisClient(RedisSideTableInfo tableInfo){
         if (database == null){
             database = "0";
         }
-        switch (tableInfo.getRedisType()){
-            case 1:
+        switch (RedisType.parse(tableInfo.getRedisType())){
+            case STANDALONE:
                 StringBuilder redisUri = new StringBuilder();
                 redisUri.append("redis://").append(password).append(url).append("/").append(database);
                 redisClient = RedisClient.create(redisUri.toString());
                 connection = redisClient.connect();
                 async = connection.async();
                 break;
-            case 2:
+            case SENTINEL:
                 StringBuilder sentinelUri = new StringBuilder();
                 sentinelUri.append("redis-sentinel://").append(password)
                         .append(url).append("/").append(database).append("#").append(redisSideTableInfo.getMasterName());
                 redisClient = RedisClient.create(sentinelUri.toString());
                 connection = redisClient.connect();
                 async = connection.async();
                 break;
-            case 3:
+            case CLUSTER:
                 StringBuilder clusterUri = new StringBuilder();
                 clusterUri.append("redis://").append(password).append(url);
                 clusterClient = RedisClusterClient.create(clusterUri.toString());
                 clusterConnection = clusterClient.connect();
                 async = clusterConnection.async();
+            default:
+                break;
         }
     }
 
diff --git a/redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/enums/RedisType.java b/redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/enums/RedisType.java
@@ -0,0 +1,33 @@
+package com.dtstack.flink.sql.side.redis.enums;
+
+public enum RedisType {
+    /**
+     * 单机
+     */
+    STANDALONE(1),
+    /**
+     * 哨兵
+     */
+    SENTINEL(2),
+    /**
+     * 集群
+     */
+    CLUSTER(3);
+    int type;
+    RedisType(int type){
+        this.type = type;
+    }
+
+    public int getType(){
+        return type;
+    }
+
+    public static RedisType parse(int redisType){
+        for(RedisType type : RedisType.values()){
+            if(type.getType() == redisType){
+                return type;
+            }
+        }
+        throw new RuntimeException("unsupport redis type["+ redisType + "]");
+    }
+}
