feat:redis store optimize

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

@@ -125,7 +125,7 @@ public class ODPSToRedis {
         log.info("argMap.maxPartitionNum: {} calcPartationNum: {}, finalPartationNum: {}", maxPartitionNum, calcPartationNum, finalPartationNum);
         log.info("config: {}", config);
         JavaRDD<Map<String, String>> repartitionedFieldValues = fieldValues.repartition(finalPartationNum);
-        repartitionedFieldValues.foreachPartition(iterator -> redisService.mSetV2(iterator, config, false));
+        repartitionedFieldValues.foreachPartition(iterator -> redisService.mSetV2(iterator, config));
 
         redisService.setMonitor(config, argMap);
 

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

@@ -15,6 +15,7 @@ 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
@@ -37,8 +38,8 @@ public class RedisService implements Serializable {
         }
     }
 
-    public void mSetV2(Iterator<Map<String, String>> dataIte, DTSConfig config, boolean isV3Optimize) {
-        log.info("mSetV2 param: {}, isV3Optimize: {}", config, isV3Optimize);
+    public void mSetV2(Iterator<Map<String, String>> dataIte, DTSConfig config) {
+        log.info("mSetV2 param: {}", config);
         Map<String, String> batch = new HashMap<>();
         Jedis jedis = new Jedis(hostName, port);
         jedis.auth(password);
@@ -61,7 +62,7 @@ public class RedisService implements Serializable {
             String value = JSONUtils.toJson(record);
             batch.put(redisKey, value);
             if (batch.size() >= 1000) {
-                mSet(jedis, batch, expire, TimeUnit.SECONDS, isV3Optimize);
+                mSet(jedis, batch, expire, TimeUnit.SECONDS);
                 batch.clear();
                 try {
                     TimeUnit.MILLISECONDS.sleep(10);
@@ -71,7 +72,7 @@ public class RedisService implements Serializable {
             }
         }
         if (MapUtils.isNotEmpty(batch)) {
-            mSet(jedis, batch, expire, TimeUnit.SECONDS, isV3Optimize);
+            mSet(jedis, batch, expire, TimeUnit.SECONDS);
         }
         jedis.close();
     }
@@ -84,20 +85,26 @@ public class RedisService implements Serializable {
         jedis.setex(redisKey, expire, "");
     }
 
-    private void mSet(Jedis jedis, Map<String, String> batch, long expire, TimeUnit timeUnit, boolean isV3Optimize) {
+    private void mSet(Jedis jedis, Map<String, String> batch, long expire, TimeUnit timeUnit) {
         long expireSeconds = timeUnit.toSeconds(expire);
 
         Pipeline pipeline = jedis.pipelined();
         for (Map.Entry<String, String> e : batch.entrySet()) {
             try {
                 String key = e.getKey();
-                if (!isV3Optimize) {
-                    key = String.format("%s:v2", e.getKey());
-                }
                 pipeline.setex(key.getBytes(StandardCharsets.UTF_8), expireSeconds, CompressionUtil.snappyCompressV2(e.getValue()));
             } catch (Exception ignore) {
             }
         }
+        try {
+            byte[][] deleteKeyArr = batch.keySet().stream()
+                    .map(s -> String.format("%s:v2", s))
+                    .map(s -> s.getBytes(StandardCharsets.UTF_8))
+                    .toArray(byte[][]::new);
+            jedis.del(deleteKeyArr);
+        } catch (Exception ignore) {
+        }
+
         pipeline.sync();
 
     }