[hotfix-34264][redis5][sink]update code ,save key and value whit jedis.hset() method.

Author: HiLany

docs/plugin/redisSink.md

@@ -44,7 +44,7 @@ redis5.0
 |minIdle|最小空闲连接数|否||0|
 |masterName| 哨兵模式下的masterName|否||
 |primarykeys|主键字段，多个字段以逗号分割|是||
-|keyExpiredTime|redis sink的key的过期时间。默认是0（永不过期），单位是ms。|否||
+|keyExpiredTime|redis sink的key的过期时间。默认是0（永不过期），单位是s。|否||
       
 
 ## 5.样例：

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

@@ -37,10 +37,9 @@
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.HashMap;
 import java.util.HashSet;
-import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -49,9 +48,17 @@
 public class RedisOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
     private static final Logger LOG = LoggerFactory.getLogger(RedisOutputFormat.class);
 
+    protected String[] fieldNames;
+
+    protected TypeInformation<?>[] fieldTypes;
+
+    protected List<String> primaryKeys;
+
+    protected int timeout = 10000;
+
     private String url;
 
-    private String database;
+    private String database = "0";
 
     private String tableName;
 
@@ -67,15 +74,7 @@ public class RedisOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
 
     private String masterName;
 
-    protected String[] fieldNames;
-
-    protected TypeInformation<?>[] fieldTypes;
-
-    protected List<String> primaryKeys;
-
-    protected int timeout;
-
-    protected long keyExpiredTime;
+    protected int keyExpiredTime;
 
     private JedisPool pool;
 
@@ -85,8 +84,13 @@ public class RedisOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
 
     private GenericObjectPoolConfig poolConfig;
 
-    private RedisOutputFormat(){
+    private RedisOutputFormat() {
     }
+
+    public static RedisOutputFormatBuilder buildRedisOutputFormat() {
+        return new RedisOutputFormatBuilder();
+    }
+
     @Override
     public void configure(Configuration parameters) {
 
@@ -98,15 +102,15 @@ public void open(int taskNumber, int numTasks) throws IOException {
         initMetric();
     }
 
-    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;
@@ -123,31 +127,23 @@ private void establishConnection() {
         for (String ipPort : nodes) {
             ipPorts.add(ipPort);
             String[] ipPortPair = StringUtils.split(ipPort, ":");
-            addresses.add(new HostAndPort(ipPortPair[0].trim(), Integer.valueOf(ipPortPair[1].trim())));
-        }
-        if (timeout == 0){
-            timeout = 10000;
-        }
-        if (database == null)
-        {
-            database = "0";
+            addresses.add(new HostAndPort(ipPortPair[0].trim(), Integer.parseInt(ipPortPair[1].trim())));
         }
 
-        switch (redisType){
-            //单机
-            case 1:
+        switch (RedisType.parse(redisType)) {
+            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(masterName, ipPorts, poolConfig, timeout, password, Integer.parseInt(database));
                 jedis = jedisSentinelPool.getResource();
                 break;
-            //集群
-            case 3:
+            case CLUSTER:
                 jedis = new JedisCluster(addresses, timeout, timeout, 10, password, poolConfig);
+                break;
             default:
+                throw new RuntimeException("unsupported redis type[ " + redisType + "]");
         }
     }
 
@@ -162,58 +158,32 @@ public void writeRecord(Tuple2 record) throws IOException {
         if (row.getArity() != fieldNames.length) {
             return;
         }
-
-        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();
-            int index = map.get(primaryKey).intValue();
-            primaryKv.append(primaryKey).append(":").append(row.getField(index));
-            kvList.add(primaryKv.toString());
-        }
-
-        String perKey = String.join(":", kvList);
+        Map<String, Object> refData = Maps.newHashMap();
         for (int i = 0; i < fieldNames.length; i++) {
-            StringBuilder key = new StringBuilder();
-            key.append(tableName).append(":").append(perKey).append(":").append(fieldNames[i]);
-
-            String value = "null";
-            Object field = row.getField(i);
-            if (field != null) {
-                value = field.toString();
-            }
-            saveKey(key.toString(), value);
+            refData.put(fieldNames[i], row.getField(i));
         }
-
-        if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0){
+        save(refData);
+        if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0) {
             LOG.info(record.toString());
         }
         outRecords.inc();
     }
 
+
     /**
-     * 1. save key and value.
-     * 2. set expired time for key when keyExpiredTime has been set.
-     * @param key
-     * @param value
+     *  1. build key from map.
+     *  2. save key and value.
+     *  3. set expired time for key when keyExpiredTime has been set.
+     * @param refData
      */
-    private void saveKey(String key, String value) {
-        if (keyExpiredTime != 0L) {
-            boolean keyExist = jedis.exists(key);
-            if (keyExist) {
-                jedis.del(key);
+    private synchronized void save(Map<String, Object> refData) {
+        String key = buildCacheKey(refData);
+        try {
+            refData.forEach((field, value) -> jedis.hset(key, field, String.valueOf(value)));
+        } finally {
+            if (keyExpiredTime != 0) {
+                jedis.expire(key, keyExpiredTime);
             }
-            jedis.set(key, value, "NX", "PX", keyExpiredTime);
-        } else {
-            jedis.set(key, value);
         }
     }
 
@@ -225,101 +195,108 @@ public void close() throws IOException {
         if (pool != null) {
             pool.close();
         }
-        if (jedis != null){
-            if (jedis instanceof Closeable){
+        if (jedis != null) {
+            if (jedis instanceof Closeable) {
                 ((Closeable) jedis).close();
             }
         }
 
     }
 
-    public static RedisOutputFormatBuilder buildRedisOutputFormat(){
-        return new RedisOutputFormatBuilder();
+    public String buildCacheKey(Map<String, Object> refData) {
+        StringBuilder keyBuilder = new StringBuilder(tableName);
+        for (String primaryKey : primaryKeys) {
+            if (!refData.containsKey(primaryKey)) {
+                return null;
+            }
+            keyBuilder.append("_").append(refData.get(primaryKey));
+        }
+        return keyBuilder.toString();
     }
 
-    public static class RedisOutputFormatBuilder{
+    public static class RedisOutputFormatBuilder {
         private final RedisOutputFormat redisOutputFormat;
 
-        protected RedisOutputFormatBuilder(){
+        protected RedisOutputFormatBuilder() {
             this.redisOutputFormat = new RedisOutputFormat();
         }
 
-        public RedisOutputFormatBuilder setUrl(String url){
+        public RedisOutputFormatBuilder setUrl(String url) {
             redisOutputFormat.url = url;
             return this;
         }
 
-        public RedisOutputFormatBuilder setDatabase(String database){
+        public RedisOutputFormatBuilder setDatabase(String database) {
             redisOutputFormat.database = database;
             return this;
         }
 
-        public RedisOutputFormatBuilder setTableName(String tableName){
+        public RedisOutputFormatBuilder setTableName(String tableName) {
             redisOutputFormat.tableName = tableName;
             return this;
         }
 
-        public RedisOutputFormatBuilder setPassword(String password){
+        public RedisOutputFormatBuilder setPassword(String password) {
             redisOutputFormat.password = password;
             return this;
         }
 
-        public RedisOutputFormatBuilder setFieldNames(String[] fieldNames){
+        public RedisOutputFormatBuilder setFieldNames(String[] fieldNames) {
             redisOutputFormat.fieldNames = fieldNames;
             return this;
         }
 
-        public RedisOutputFormatBuilder setFieldTypes(TypeInformation<?>[] fieldTypes){
+        public RedisOutputFormatBuilder setFieldTypes(TypeInformation<?>[] fieldTypes) {
             redisOutputFormat.fieldTypes = fieldTypes;
             return this;
         }
 
-        public RedisOutputFormatBuilder setPrimaryKeys(List<String > primaryKeys){
+        public RedisOutputFormatBuilder setPrimaryKeys(List<String> primaryKeys) {
             redisOutputFormat.primaryKeys = primaryKeys;
             return this;
         }
 
-        public RedisOutputFormatBuilder setTimeout(int timeout){
+        public RedisOutputFormatBuilder setTimeout(int timeout) {
             redisOutputFormat.timeout = timeout;
             return this;
         }
 
-        public RedisOutputFormatBuilder setRedisType(int redisType){
+        public RedisOutputFormatBuilder setRedisType(int redisType) {
             redisOutputFormat.redisType = redisType;
             return this;
         }
 
-        public RedisOutputFormatBuilder setMaxTotal(String maxTotal){
+        public RedisOutputFormatBuilder setMaxTotal(String maxTotal) {
             redisOutputFormat.maxTotal = maxTotal;
             return this;
         }
 
-        public RedisOutputFormatBuilder setMaxIdle(String maxIdle){
+        public RedisOutputFormatBuilder setMaxIdle(String maxIdle) {
             redisOutputFormat.maxIdle = maxIdle;
             return this;
         }
 
-        public RedisOutputFormatBuilder setMinIdle(String minIdle){
+        public RedisOutputFormatBuilder setMinIdle(String minIdle) {
             redisOutputFormat.minIdle = minIdle;
             return this;
         }
 
-        public RedisOutputFormatBuilder setMasterName(String masterName){
+        public RedisOutputFormatBuilder setMasterName(String masterName) {
             redisOutputFormat.masterName = masterName;
             return this;
         }
 
-        public RedisOutputFormatBuilder setKeyExpiredTime(long keyExpiredTime){
+        public RedisOutputFormatBuilder setKeyExpiredTime(int keyExpiredTime){
             redisOutputFormat.keyExpiredTime = keyExpiredTime;
             return this;
         }
 
-        public RedisOutputFormat finish(){
-            if (redisOutputFormat.url == null){
+        public RedisOutputFormat finish() {
+            if (redisOutputFormat.url == null) {
                 throw new IllegalArgumentException("No URL supplied.");
             }
 
-            if (redisOutputFormat.tableName == null){
+            if (redisOutputFormat.tableName == null) {
                 throw new IllegalArgumentException("No tablename supplied.");
             }
 

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

@@ -71,9 +71,9 @@ public class RedisSink implements RetractStreamTableSink<Row>, IStreamSinkGener<
 
     protected String registerTableName;
 
-    protected long keyExpiredTime;
+    protected int keyExpiredTime;
 
-    public RedisSink(){
+    public RedisSink() {
 
     }
 
@@ -111,7 +111,7 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
     @Override
     public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         RedisOutputFormat.RedisOutputFormatBuilder builder = RedisOutputFormat.buildRedisOutputFormat();
-        builder.setUrl(this.url)
+        RedisOutputFormat redisOutputFormat = builder.setUrl(this.url)
                 .setDatabase(this.database)
                 .setTableName(this.tableName)
                 .setPassword(this.password)
@@ -124,8 +124,8 @@ public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<
                 .setMaxIdle(this.maxIdle)
                 .setMinIdle(this.minIdle)
                 .setMasterName(this.masterName)
-                .setKeyExpiredTime(this.keyExpiredTime);
-        RedisOutputFormat redisOutputFormat = builder.finish();
+                .setKeyExpiredTime(this.keyExpiredTime)
+                .finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(redisOutputFormat);
         DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction)
                 .setParallelism(parallelism)

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

@@ -28,6 +28,6 @@ public static RedisType parse(int redisType){
                 return type;
             }
         }
-        throw new RuntimeException("unsupport redis type["+ redisType + "]");
+        throw new RuntimeException("unsupported redis type["+ redisType + "]");
     }
 }

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

@@ -50,7 +50,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         redisTableInfo.setMasterName(MathUtil.getString(props.get(RedisTableInfo.MASTER_NAME.toLowerCase())));
 
         if (props.get(RedisTableInfo.KEY_EXPIRED_TIME.toLowerCase()) != null) {
-            redisTableInfo.setKeyExpiredTime(Long.parseLong(MathUtil.getString(props.get(RedisTableInfo.KEY_EXPIRED_TIME.toLowerCase()))));
+            redisTableInfo.setKeyExpiredTime(Integer.parseInt(MathUtil.getString(props.get(RedisTableInfo.KEY_EXPIRED_TIME.toLowerCase()))));
         }
 
         String primaryKeysStr = MathUtil.getString(props.get(RedisTableInfo.PRIMARY_KEYS_NAME));