[feat-34267][redis] opt code.

FlechazoW · FlechazoW · commit 9281216f998d · 2021-02-04T20:50:03.000+08:00
diff --git a/redis5/pom.xml b/redis5/pom.xml
@@ -15,6 +15,18 @@
         <module>redis5-sink</module>
         <module>redis5-side</module>
     </modules>
-
+    <dependencies>
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.core</artifactId>
+            <version>1.0-SNAPSHOT</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>redis.clients</groupId>
+            <artifactId>jedis</artifactId>
+            <version>2.9.0</version>
+        </dependency>
+    </dependencies>
     
 </project>
diff --git a/redis5/redis5-side/pom.xml b/redis5/redis5-side/pom.xml
@@ -11,20 +11,6 @@
     <modelVersion>4.0.0</modelVersion>
     <artifactId>sql.side.redis</artifactId>
     <name>redis-side</name>
-    <dependencies>
-        <dependency>
-            <groupId>com.dtstack.flink</groupId>
-            <artifactId>sql.core</artifactId>
-            <version>1.0-SNAPSHOT</version>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>redis.clients</groupId>
-            <artifactId>jedis</artifactId>
-            <version>2.8.0</version>
-        </dependency>
-    </dependencies>
-
 
     <modules>
         <module>redis-side-core</module>
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
@@ -30,7 +30,6 @@
 import com.google.common.collect.Maps;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.table.dataformat.BaseRow;
@@ -48,10 +47,12 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.sql.SQLException;
+import java.util.Arrays;
 import java.util.Calendar;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.atomic.AtomicReference;
@@ -117,9 +118,9 @@ protected void reloadCache() {
     public void flatMap(BaseRow input, Collector<BaseRow> out) throws Exception {
         GenericRow genericRow = (GenericRow) input;
         Map<String, Object> inputParams = Maps.newHashMap();
-        for(Integer conValIndex : sideInfo.getEqualValIndex()){
+        for (Integer conValIndex : sideInfo.getEqualValIndex()) {
             Object equalObj = genericRow.getField(conValIndex);
-            if(equalObj == null){
+            if (equalObj == null) {
                 if (sideInfo.getJoinType() == JoinType.LEFT) {
                     BaseRow data = fillData(input, null);
                     RowDataComplete.collectBaseRow(out, data);
@@ -133,11 +134,11 @@ public void flatMap(BaseRow input, Collector<BaseRow> out) throws Exception {
 
         Map<String, String> cacheMap = cacheRef.get().get(key);
 
-        if (cacheMap == null){
-            if(sideInfo.getJoinType() == JoinType.LEFT){
+        if (cacheMap == null) {
+            if (sideInfo.getJoinType() == JoinType.LEFT) {
                 BaseRow data = fillData(input, null);
                 RowDataComplete.collectBaseRow(out, data);
-            }else{
+            } else {
                 return;
             }
 
@@ -187,49 +188,54 @@ private JedisCommands getJedis(RedisSideTableInfo tableInfo) {
         String url = tableInfo.getUrl();
         String password = tableInfo.getPassword();
         String database = tableInfo.getDatabase() == null ? "0" : tableInfo.getDatabase();
+        String masterName = tableInfo.getMasterName();
         int timeout = tableInfo.getTimeout();
-        if (timeout == 0){
-            timeout = 1000;
-        }
-
-        String[] nodes = StringUtils.split(url, ",");
-        String[] firstIpPort = StringUtils.split(nodes[0], ":");
-        String firstIp = firstIpPort[0];
-        String firstPort = firstIpPort[1];
-        Set<HostAndPort> addresses = new HashSet<>();
-        Set<String> ipPorts = new HashSet<>();
-
-        // 对ipv6 支持
-        for (String node : nodes) {
-            ipPorts.add(node);
-            Matcher matcher = HOST_PORT_PATTERN.matcher(node);
-            if (matcher.find()) {
-                String host = matcher.group("host").trim();
-                String portStr = matcher.group("port").trim();
-                if (StringUtils.isNotBlank(host) && StringUtils.isNotBlank(portStr)) {
-                    // 转化为int格式的端口
-                    int port = Integer.parseInt(portStr);
-                    addresses.add(new HostAndPort(host, port));
-                }
-            }
+        if (timeout == 0) {
+            timeout = 10000;
         }
 
+        String[] nodes = url.split(",");
         JedisCommands jedis = null;
         GenericObjectPoolConfig poolConfig = setPoolConfig(tableInfo.getMaxTotal(), tableInfo.getMaxIdle(), tableInfo.getMinIdle());
-        switch (RedisType.parse(tableInfo.getRedisType())){
-            //单机
+        switch (RedisType.parse(tableInfo.getRedisType())) {
             case STANDALONE:
-                pool = new JedisPool(poolConfig, firstIp, Integer.parseInt(firstPort), timeout, password, Integer.parseInt(database));
-                jedis = pool.getResource();
+                String firstIp = null;
+                String firstPort = null;
+                Matcher standalone = HOST_PORT_PATTERN.matcher(nodes[0]);
+                if (standalone.find()) {
+                    firstIp = standalone.group("host").trim();
+                    firstPort = standalone.group("port").trim();
+                }
+                if (Objects.nonNull(firstIp)) {
+                    pool = new JedisPool(poolConfig, firstIp, Integer.parseInt(firstPort), timeout, password, Integer.parseInt(database));
+                    jedis = pool.getResource();
+                } else {
+                    throw new IllegalArgumentException(
+                            String.format("redis url error. current url [%s]", nodes[0]));
+                }
                 break;
-            //哨兵
             case SENTINEL:
-                jedisSentinelPool = new JedisSentinelPool(tableInfo.getMasterName(), ipPorts, poolConfig, timeout, password, Integer.parseInt(database));
+                Set<String> ipPorts = new HashSet<>(Arrays.asList(nodes));
+                jedisSentinelPool = new JedisSentinelPool(masterName, ipPorts, poolConfig, timeout, password, Integer.parseInt(database));
                 jedis = jedisSentinelPool.getResource();
                 break;
-            //集群
             case CLUSTER:
-                jedis = new JedisCluster(addresses, timeout, timeout,1, poolConfig);
+                Set<HostAndPort> addresses = new HashSet<>();
+                // 对ipv6 支持
+                for (String node : nodes) {
+                    Matcher matcher = HOST_PORT_PATTERN.matcher(node);
+                    if (matcher.find()) {
+                        String host = matcher.group("host").trim();
+                        String portStr = matcher.group("port").trim();
+                        if (org.apache.commons.lang3.StringUtils.isNotBlank(host) && org.apache.commons.lang3.StringUtils.isNotBlank(portStr)) {
+                            // 转化为int格式的端口
+                            int port = Integer.parseInt(portStr);
+                            addresses.add(new HostAndPort(host, port));
+                        }
+                    }
+                }
+                jedis = new JedisCluster(addresses, timeout, timeout, 10, password, poolConfig);
+                break;
             default:
                 break;
         }
@@ -257,35 +263,32 @@ private JedisCommands getJedisWithRetry(int retryNum) {
         return null;
     }
 
-    private Set<String> getRedisKeys(RedisType redisType, JedisCommands jedis, String keyPattern){
-        if(!redisType.equals(RedisType.CLUSTER)){
+    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)jedis).getClusterNodes();
-        for(String k : clusterNodes.keySet()){
+        Map<String, JedisPool> clusterNodes = ((JedisCluster) jedis).getClusterNodes();
+        for (String k : clusterNodes.keySet()) {
             JedisPool jp = clusterNodes.get(k);
-            Jedis connection = jp.getResource();
-            try {
+            try (Jedis connection = jp.getResource()) {
                 keys.addAll(connection.keys(keyPattern));
-            } catch (Exception e){
-                LOG.error("Getting keys error: {}", e);
-            } finally {
-                connection.close();
+            } catch (Exception e) {
+                LOG.error("Getting keys error", e);
             }
         }
         return keys;
     }
 
-    private GenericObjectPoolConfig setPoolConfig(String maxTotal, String maxIdle, String minIdle){
+    private GenericObjectPoolConfig setPoolConfig(String maxTotal, String maxIdle, String minIdle) {
         GenericObjectPoolConfig config = new GenericObjectPoolConfig();
-        if (maxTotal != null){
+        if (maxTotal != null) {
             config.setMaxTotal(Integer.parseInt(maxTotal));
         }
-        if (maxIdle != null){
+        if (maxIdle != null) {
             config.setMaxIdle(Integer.parseInt(maxIdle));
         }
-        if (minIdle != null){
+        if (minIdle != null) {
             config.setMinIdle(Integer.parseInt(minIdle));
         }
         return config;
diff --git a/redis5/redis5-side/redis-side-core/pom.xml b/redis5/redis5-side/redis-side-core/pom.xml
@@ -12,14 +12,7 @@
     </parent>
 
     <artifactId>sql.side.redis.core</artifactId>
-    <dependencies>
-        <dependency>
-            <groupId>com.dtstack.flink</groupId>
-            <artifactId>sql.core</artifactId>
-            <version>1.0-SNAPSHOT</version>
-            <scope>provided</scope>
-        </dependency>
-    </dependencies>
+
     <packaging>jar</packaging>
 
     
diff --git a/redis5/redis5-sink/pom.xml b/redis5/redis5-sink/pom.xml
@@ -16,20 +16,6 @@
     <name>redis-sink</name>
     <url>http://maven.apache.org</url>
 
-    <dependencies>
-        <dependency>
-            <groupId>com.dtstack.flink</groupId>
-            <artifactId>sql.core</artifactId>
-            <version>1.0-SNAPSHOT</version>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>redis.clients</groupId>
-            <artifactId>jedis</artifactId>
-            <version>2.9.0</version>
-        </dependency>
-    </dependencies>
-
     <build>
         <plugins>
             <plugin>
diff --git a/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java b/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java
@@ -37,9 +37,11 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -121,41 +123,68 @@ private GenericObjectPoolConfig setPoolConfig(String maxTotal, String maxIdle, S
     private void establishConnection() {
         GenericObjectPoolConfig poolConfig = setPoolConfig(maxTotal, maxIdle, minIdle);
         String[] nodes = StringUtils.split(url, ",");
-        String[] firstIpPort = StringUtils.split(nodes[0], ":");
-        String firstIp = firstIpPort[0];
-        String firstPort = firstIpPort[1];
-        Set<HostAndPort> addresses = new HashSet<>();
-        Set<String> ipPorts = new HashSet<>();
-
-        // 对ipv6 支持
-        for (String node : nodes) {
-            ipPorts.add(node);
-            Matcher matcher = HOST_PORT_PATTERN.matcher(node);
-            if (matcher.find()) {
-                String host = matcher.group("host").trim();
-                String portStr = matcher.group("port").trim();
-                if (StringUtils.isNotBlank(host) && StringUtils.isNotBlank(portStr)) {
-                    // 转化为int格式的端口
-                    int port = Integer.parseInt(portStr);
-                    addresses.add(new HostAndPort(host, port));
-                }
-            }
-        }
 
-        switch (RedisType.parse(redisType)) {
+        switch (RedisType.parse(redisType)){
             case STANDALONE:
-                pool = new JedisPool(poolConfig, firstIp, Integer.parseInt(firstPort), timeout, password, Integer.parseInt(database));
-                jedis = pool.getResource();
+                String firstIp = null;
+                String firstPort = null;
+                Matcher standalone = HOST_PORT_PATTERN.matcher(nodes[0]);
+                if (standalone.find()) {
+                    firstIp = standalone.group("host").trim();
+                    firstPort = standalone.group("port").trim();
+                }
+                if (Objects.nonNull(firstIp)) {
+                    pool = new JedisPool(
+                            poolConfig,
+                            firstIp,
+                            Integer.parseInt(firstPort),
+                            timeout,
+                            password,
+                            Integer.parseInt(database));
+
+                    jedis = pool.getResource();
+                } else {
+                    throw new IllegalArgumentException(
+                            String.format("redis url error. current url [%s]", nodes[0]));
+                }
                 break;
             case SENTINEL:
-                jedisSentinelPool = new JedisSentinelPool(masterName, ipPorts, poolConfig, timeout, password, Integer.parseInt(database));
+                Set<String> ipPorts = new HashSet<>(Arrays.asList(nodes));
+                jedisSentinelPool = new JedisSentinelPool(
+                        masterName,
+                        ipPorts,
+                        poolConfig,
+                        timeout,
+                        password,
+                        Integer.parseInt(database));
+
                 jedis = jedisSentinelPool.getResource();
                 break;
             case CLUSTER:
-                jedis = new JedisCluster(addresses, timeout, timeout, 10, password, poolConfig);
+                Set<HostAndPort> addresses = new HashSet<>();
+                // 对ipv6 支持
+                for (String node : nodes) {
+                    Matcher matcher = HOST_PORT_PATTERN.matcher(node);
+                    if (matcher.find()) {
+                        String host = matcher.group("host").trim();
+                        String portStr = matcher.group("port").trim();
+                        if (StringUtils.isNotBlank(host) && StringUtils.isNotBlank(portStr)) {
+                            // 转化为int格式的端口
+                            int port = Integer.parseInt(portStr);
+                            addresses.add(new HostAndPort(host, port));
+                        }
+                    }
+                }
+                jedis = new JedisCluster(
+                        addresses,
+                        timeout,
+                        timeout,
+                        10,
+                        password,
+                        poolConfig);
                 break;
             default:
-                throw new RuntimeException("unsupported redis type[ " + redisType + "]");
+                throw new IllegalArgumentException("unsupported redis type[ " + redisType + "]");
         }
     }
 
