redis can

Author: todd5167

redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java

@@ -39,7 +39,9 @@
 import java.util.*;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
-
+/**
+ * @author yanxi
+ */
 public class RedisAllReqRow extends AllReqRow{
 
     private static final long serialVersionUID = 7578879189085344807L;
@@ -259,7 +261,8 @@ private JedisCommands getJedis(RedisSideTableInfo tableInfo) {
                 break;
             //集群
             case 3:
-                jedis = new JedisCluster(addresses, timeout, timeout,1, poolConfig);
+                jedis = new JedisCluster(addresses, timeout, timeout, 1, poolConfig);
+            default:
         }
 
         return jedis;

redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllSideInfo.java

@@ -28,7 +28,9 @@
 import com.google.common.collect.Lists;
 
 import java.util.List;
-
+/**
+ * @author yanxi
+ */
 public class RedisAllSideInfo extends SideInfo {
 
     private static final long serialVersionUID = 1998703966487857613L;

redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java

@@ -43,7 +43,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
-
+/**
+ * @author yanxi
+ */
 public class RedisAsyncReqRow extends AsyncReqRow {
 
     private static final long serialVersionUID = -2079908694523987738L;
@@ -108,6 +110,7 @@ private void buildRedisClient(RedisSideTableInfo tableInfo){
                 clusterClient = RedisClusterClient.create(clusterUri.toString());
                 clusterConnection = clusterClient.connect();
                 async = clusterConnection.async();
+            default:
         }
     }
 

redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncSideInfo.java

@@ -28,7 +28,9 @@
 import com.google.common.collect.Lists;
 
 import java.util.List;
-
+/**
+ * @author yanxi
+ */
 public class RedisAsyncSideInfo extends SideInfo {
     private static final long serialVersionUID = -4851348392924455039L;
 

redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/table/RedisSideParser.java

@@ -24,6 +24,9 @@
 
 import java.util.Map;
 
+/**
+ * @author yanxi
+ */
 public class RedisSideParser extends AbsSideTableParser {
 
     @Override

redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/table/RedisSideTableInfo.java

@@ -20,7 +20,9 @@
 
 import com.dtstack.flink.sql.side.SideTableInfo;
 import com.google.common.base.Preconditions;
-
+/**
+ * @author yanxi
+ */
 public class RedisSideTableInfo extends SideTableInfo {
 
     private static final long serialVersionUID = -1L;

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java

@@ -26,11 +26,23 @@
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import redis.clients.jedis.*;
+import redis.clients.jedis.HostAndPort;
+import redis.clients.jedis.JedisCluster;
+import redis.clients.jedis.JedisCommands;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.JedisSentinelPool;
+
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.*;
-
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * @author yanxi
+ */
 public class RedisOutputFormat extends DtRichOutputFormat<Tuple2> {
     private static final Logger LOG = LoggerFactory.getLogger(RedisOutputFormat.class);
 
@@ -129,7 +141,8 @@ private void establishConnection() {
                 break;
             //集群
             case 3:
-                jedis = new JedisCluster(addresses, timeout, timeout,10, password, poolConfig);
+                jedis = new JedisCluster(addresses, timeout, timeout, 10, password, poolConfig);
+            default:
         }
     }
 
@@ -145,21 +158,21 @@ public void writeRecord(Tuple2 record) throws IOException {
             return;
         }
 
-        HashMap<String, Integer> map = new HashMap<>();
-        for (String primaryKey : primaryKeys){
-            for (int i=0; i<fieldNames.length; i++){
-                if (fieldNames[i].equals(primaryKey)){
+        HashMap<String, Integer> map = new HashMap<>(8);
+        for (String primaryKey : primaryKeys) {
+            for (int i = 0; i < fieldNames.length; i++) {
+                if (fieldNames[i].equals(primaryKey)) {
                     map.put(primaryKey, i);
                 }
             }
         }
 
         List<String> kvList = new LinkedList<>();
         for (String primaryKey : primaryKeys){
-            StringBuilder primaryKV = new StringBuilder();
+            StringBuilder primaryKv = new StringBuilder();
             int index = map.get(primaryKey).intValue();
-            primaryKV.append(primaryKey).append(":").append(row.getField(index));
-            kvList.add(primaryKV.toString());
+            primaryKv.append(primaryKey).append(":").append(row.getField(index));
+            kvList.add(primaryKv.toString());
         }
 
         String perKey = String.join(":", kvList);

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisSink.java

@@ -33,7 +33,9 @@
 import org.apache.flink.types.Row;
 
 import java.util.List;
-
+/**
+ * @author yanxi
+ */
 public class RedisSink implements RetractStreamTableSink<Row>, IStreamSinkGener<RedisSink> {
 
     protected String[] fieldNames;

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/table/RedisSinkParser.java

@@ -24,11 +24,12 @@
 import com.google.common.collect.Lists;
 import org.apache.commons.lang3.StringUtils;
 
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-
+/**
+ * @author yanxi
+ */
 public class RedisSinkParser extends AbsTableParser {
     @Override
     public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/table/RedisTableInfo.java

@@ -20,7 +20,9 @@
 
 import com.dtstack.flink.sql.table.TargetTableInfo;
 import com.google.common.base.Preconditions;
-
+/**
+ * @author yanxi
+ */
 public class RedisTableInfo extends TargetTableInfo {
 
     private static final String CURR_TYPE = "redis";