feat:redis store optimize

recommend-feature-produce/src/main/java/com/tzld/piaoquan/recommend/feature/produce/service/RedisService.java

@@ -15,7 +15,6 @@ import java.io.Serializable;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
 
 /**
  * https://help.aliyun.com/zh/maxcompute/user-guide/java-sdk-1/?spm=a2c4g.11174283.0.0.6d0111c1E15lI3
@@ -46,20 +45,11 @@ public class RedisService implements Serializable {
         long expire = config.getRedis().getExpire();
         while (dataIte.hasNext()) {
             Map<String, String> record = dataIte.next();
-            String redisKey = redisKey(record, config);
 
-            Map<String, String> valueMap = new HashMap<>();
-            if (StringUtils.isNotBlank(config.getRedis().getFeatureField())) {
-                record = JSONUtils.fromJson(record.get(config.getRedis().getFeatureField()), new TypeToken<Map<String, String>>() {
-                }, Collections.emptyMap());
-            }else if (CollectionUtils.isNotEmpty(config.getRedis().getValue())) {
-                for (String valueKey : config.getRedis().getValue()) {
-                    valueMap.put(valueKey, record.get(valueKey));
-                }
-                record = valueMap;
-            }
+            String redisKey = buildRedisKey(record, config);
+            Map<String, String> valueMap = this.buildValueMap(record, config);
 
-            String value = JSONUtils.toJson(record);
+            String value = JSONUtils.toJson(valueMap);
             batch.put(redisKey, value);
             if (batch.size() >= 1000) {
                 mSet(jedis, batch, expire, TimeUnit.SECONDS);
@@ -78,11 +68,13 @@ public class RedisService implements Serializable {
     }
 
     public void setMonitor(DTSConfig config, Map<String, String> argMap) {
-        Jedis jedis = new Jedis(hostName, port);
-        jedis.auth(password);
-        long expire = config.getRedis().getExpire();
-        String redisKey = "ttl:" + argMap.get("table");
-        jedis.setex(redisKey, expire, "");
+        try (Jedis jedis = new Jedis(hostName, port)) {
+            jedis.auth(password);
+            long expire = config.getRedis().getExpire();
+            String redisKey = "ttl:" + argMap.get("table");
+            jedis.setex(redisKey, expire, "");
+        } catch (Exception ignore) {
+        }
     }
 
     private void mSet(Jedis jedis, Map<String, String> batch, long expire, TimeUnit timeUnit) {
@@ -91,25 +83,16 @@ public class RedisService implements Serializable {
         Pipeline pipeline = jedis.pipelined();
         for (Map.Entry<String, String> e : batch.entrySet()) {
             try {
-                String key = e.getKey();
-                pipeline.setex(key.getBytes(StandardCharsets.UTF_8), expireSeconds, CompressionUtil.snappyCompressV2(e.getValue()));
+                pipeline.setex(e.getKey().getBytes(StandardCharsets.UTF_8), expireSeconds, CompressionUtil.snappyCompressV2(e.getValue()));
             } catch (Exception ignore) {
             }
-
-            try {
-                String deleteKey = String.format("%s:v2", e.getKey());
-                pipeline.del(deleteKey.getBytes(StandardCharsets.UTF_8));
-            }catch (Exception ignore){
-
-            }
-
         }
 
         pipeline.sync();
 
     }
 
-    public String redisKey(Map<String, String> fieldValueMap, DTSConfig config) {
+    public String buildRedisKey(Map<String, String> fieldValueMap, DTSConfig config) {
         // Note：写入和读取的key生成规则应保持一致
         List<String> fields = config.getRedis().getKey();
         if (CollectionUtils.isEmpty(fields)) {
@@ -125,4 +108,19 @@ public class RedisService implements Serializable {
         }
         return sb.toString();
     }
+
+    public Map<String, String> buildValueMap(Map<String, String> record, DTSConfig config) {
+        Map<String, String> valueMap = new HashMap<>();
+        if (StringUtils.isNotBlank(config.getRedis().getFeatureField())) {
+            valueMap = JSONUtils.fromJson(record.get(config.getRedis().getFeatureField()), new TypeToken<Map<String, String>>() {
+            }, Collections.emptyMap());
+        } else if (CollectionUtils.isNotEmpty(config.getRedis().getValue())) {
+            for (String valueKey : config.getRedis().getValue()) {
+                valueMap.put(valueKey, record.get(valueKey));
+            }
+        } else {
+            valueMap = record;
+        }
+        return valueMap;
+    }
 }