JedisFlowClient uses jedis pool (#23)

weibocom · Dec 13, 2023 · 299002d · 299002d
1 parent 0f95442
commit 299002d
Show file tree

Hide file tree

Showing 7 changed files with 187 additions and 169 deletions.
diff --git a/...e-storage/src/main/java/com/weibo/rill/flow/olympicene/storage/redis/api/RedisClient.java b/...e-storage/src/main/java/com/weibo/rill/flow/olympicene/storage/redis/api/RedisClient.java
@@ -115,4 +115,8 @@ public interface RedisClient {
     Object eval(String script, String shardingKey, List<String> keys, List<String> args);
     long incr(String key);
     long hset(String key, String field, String value);
+
+    Set<String> hkeys(String key);
+
+    String hget(String key, String field);
 }
diff --git a/...main/java/com/weibo/rill/flow/olympicene/storage/redis/apicommons/GroupedRedisClient.java b/...main/java/com/weibo/rill/flow/olympicene/storage/redis/apicommons/GroupedRedisClient.java
@@ -258,7 +258,18 @@ public Object eval(String script, String shardingKey, List<String> keys, List<St
         return choose(shardingKey).eval(script, shardingKey, keys, args);
     }
 
+    @Override
+    public Set<String> hkeys(String key) {
+        return choose(key).hkeys(key);
+    }
+
+    @Override
+    public String hget(String key, String field) {
+        return choose(key).hget(key, field);
+    }
+
     private RedisClient choose(byte[] shardingKey) {
         return choose(new String(shardingKey, StandardCharsets.UTF_8));
     }
+
 }
diff --git a/rill-flow-impl/src/main/java/com/weibo/rill/flow/impl/redis/CustomizedStorageImpl.java b/rill-flow-impl/src/main/java/com/weibo/rill/flow/impl/redis/CustomizedStorageImpl.java
@@ -50,14 +50,14 @@ public String initBucket(String bucketName, JSONObject fieldToValues) {
         int reserveTimeInSecond = getReserveTimeInSecond(bucketName);
 
         JedisFlowClient jedisFlowClient = getJedisClient(bucketName);
-        try (Pipeline pipeline = jedisFlowClient.pipelined()) {
+        jedisFlowClient.pipelined().accept(pipeline -> {
             pipeline.hset(bucketName, "expire", String.valueOf(currentTimeSecond + reserveTimeInSecond));
             if (MapUtils.isNotEmpty(fieldToValues)) {
                 fieldToValues.forEach((field, value) -> pipeline.hset(bucketName, field, value.toString()));
             }
             pipeline.expire(bucketName, reserveTimeInSecond);
             pipeline.sync();
-        }
+        });
         return bucketName;
     }
 
@@ -67,10 +67,10 @@ public void store(String bucketName, JSONObject fieldToValues) {
         }
 
         JedisFlowClient jedisFlowClient = getJedisClient(bucketName);
-        try (Pipeline pipeline = jedisFlowClient.pipelined()) {
+        jedisFlowClient.pipelined().accept(pipeline -> {
             fieldToValues.forEach((field, value) -> pipeline.hset(bucketName, field, value.toString()));
             pipeline.sync();
-        }
+        });
     }
 
     public Map<String, String> load(String bucketName, boolean hGetAll, List<String> fieldNames, String fieldPrefix) {
@@ -105,9 +105,7 @@ public boolean remove(String bucketName) {
     public boolean remove(String bucketName, List<String> fieldNames) {
         JedisFlowClient jedisFlowClient = getJedisClient(bucketName);
         if (CollectionUtils.isNotEmpty(fieldNames)) {
-            try (Pipeline pipeline = jedisFlowClient.pipelined()) {
-                fieldNames.forEach(fieldName -> pipeline.hdel(bucketName, fieldName));
-            }
+            jedisFlowClient.pipelined().accept(pipeline -> fieldNames.forEach(fieldName -> pipeline.hdel(bucketName, fieldName)));
         }
         return true;
     }