feat(kv): zset version1 encode support async write (#348)

netease-im · Nov 8, 2024 · f685ffc · f685ffc
1 parent 6aea084
commit f685ffc
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 2 deletions.
diff --git a/...ore/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/kv/command/Commander.java b/...ore/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/kv/command/Commander.java
@@ -104,4 +104,8 @@ protected final void submitAsyncWriteTask(int slot, Result result, Runnable runn
             throw e;
         }
     }
+
+    protected final void submitAsyncWriteTask(int slot, Runnable runnable) {
+        asyncWriteExecutor.submit(slot, runnable);
+    }
 }
diff --git a/...ain/java/com/netease/nim/camellia/redis/proxy/upstream/kv/command/zset/ZAddCommander.java b/...ain/java/com/netease/nim/camellia/redis/proxy/upstream/kv/command/zset/ZAddCommander.java
@@ -279,7 +279,11 @@ private Reply zaddVersion1(int slot, KeyMeta keyMeta, byte[] key, byte[] cacheKe
         if (result.isKvWriteDelayEnable()) {
             submitAsyncWriteTask(slot, result, () -> kvClient.batchPut(slot, list));
         } else {
-            kvClient.batchPut(slot, list);
+            if (cacheConfig.isZSetVersion1KvWriteAsyncEnable()) {
+                submitAsyncWriteTask(slot, () -> kvClient.batchPut(slot, list));
+            } else {
+                kvClient.batchPut(slot, list);
+            }
         }
         List<CompletableFuture<Reply>> futureList = null;
         if (!memberIndexCacheWriteCommands.isEmpty()) {

diff --git a/...in/java/com/netease/nim/camellia/redis/proxy/upstream/kv/command/zset/ZRem0Commander.java b/...in/java/com/netease/nim/camellia/redis/proxy/upstream/kv/command/zset/ZRem0Commander.java
@@ -151,7 +151,11 @@ protected final Reply zremVersion1(int slot, KeyMeta keyMeta, byte[] key, byte[]
             });
         } else {
             if (!deleteSubKeys.isEmpty()) {
-                kvClient.batchDelete(slot, deleteSubKeys.toArray(new byte[0][0]));
+                if (cacheConfig.isZSetVersion1KvWriteAsyncEnable()) {
+                    submitAsyncWriteTask(slot, () -> kvClient.batchDelete(slot, deleteSubKeys.toArray(new byte[0][0])));
+                } else {
+                    kvClient.batchDelete(slot, deleteSubKeys.toArray(new byte[0][0]));
+                }
             }
         }
 

diff --git a/...re/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/kv/domain/CacheConfig.java b/...re/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/kv/domain/CacheConfig.java
@@ -33,6 +33,8 @@ public class CacheConfig {
     private final SetLRUCache setLRUCache;
     private final ZSetIndexLRUCache zSetIndexLRUCache;
 
+    private boolean zsetVersion1KvWriteAsyncEnable;
+
     public CacheConfig(String namespace) {
         this.namespace = namespace;
         initCacheConfig();
@@ -117,6 +119,11 @@ private void initCacheConfig() {
             }
             logger.info("kv.set.local.cache.enable = {}, namespace = {}", setLocalCacheEnable, namespace);
         }
+        boolean zsetVersion1KvWriteAsyncEnable = RedisKvConf.getBoolean(namespace, "kv.zset.version1.kv.async.write.enable", false);
+        if ((this.zsetVersion1KvWriteAsyncEnable && !zsetVersion1KvWriteAsyncEnable) || (!this.zsetVersion1KvWriteAsyncEnable && zsetVersion1KvWriteAsyncEnable)) {
+            this.zsetVersion1KvWriteAsyncEnable = zsetVersion1KvWriteAsyncEnable;
+            logger.info("kv.zset.version1.kv.async.write.enable = {}, namespace = {}",zsetVersion1KvWriteAsyncEnable, namespace);
+        }
     }
 
     public String getNamespace() {
@@ -139,6 +146,10 @@ public boolean isSetLocalCacheEnable() {
         return setLocalCacheEnable;
     }
 
+    public boolean isZSetVersion1KvWriteAsyncEnable() {
+        return zsetVersion1KvWriteAsyncEnable;
+    }
+
     public KeyMetaLRUCache getKeyMetaLRUCache() {
         return keyMetaLRUCache;
     }