From 2e0d9af014df86cdf9ef45db56e11e24111e4346 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com>
Date: Wed, 25 Oct 2023 17:50:31 +0800
Subject: [PATCH] =?UTF-8?q?=E5=90=8C=E6=AD=A5=EF=BC=9Aredkale=20=E5=8D=87?=
=?UTF-8?q?=E7=BA=A7?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
pom.xml | 8 +-
src/main/java/net/tccn/RedisTest.java | 137 +-
src/main/java/net/tccn/ZhubAgentProvider.java | 5 -
.../cache/redis/AbstractRedisSource.java | 131 +-
.../cache/redis/MyRedisCacheSource.java | 59 +-
.../org/redkalex/cache/redis/RedisCRC16.java | 55 +-
.../cache/redis/RedisCacheClient.java | 40 +-
.../redkalex/cache/redis/RedisCacheCodec.java | 266 +--
.../cache/redis/RedisCacheConnection.java | 19 +-
.../cache/redis/RedisCacheReqAuth.java | 23 +-
.../cache/redis/RedisCacheReqClientName.java | 45 +
.../cache/redis/RedisCacheReqClose.java | 25 +-
.../redkalex/cache/redis/RedisCacheReqDB.java | 23 +-
.../cache/redis/RedisCacheReqPing.java | 25 +-
.../cache/redis/RedisCacheRequest.java | 93 +-
.../cache/redis/RedisCacheResult.java | 173 +-
.../cache/redis/RedisCacheSource.java | 1878 +++++++----------
.../cache/redis/RedisCacheSourceProvider.java | 7 +-
.../redkalex/cache/redis/RedisCommand.java | 124 ++
.../org/redkalex/cache/redis/RedisConfig.java | 190 ++
.../redkalex/cache/redis/RedisCryptor.java | 3 +
21 files changed, 1783 insertions(+), 1546 deletions(-)
create mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheReqClientName.java
create mode 100644 src/main/java/org/redkalex/cache/redis/RedisCommand.java
create mode 100644 src/main/java/org/redkalex/cache/redis/RedisConfig.java
diff --git a/pom.xml b/pom.xml
index 79bb1b4..9c15025 100644
--- a/pom.xml
+++ b/pom.xml
@@ -6,7 +6,7 @@
net.tccn
zhub-client-redkale
- 0.1.2-dev
+ 0.1.1-dev
17
@@ -21,6 +21,12 @@
2.8.0-dev
compile
+
+ junit
+ junit
+ 4.13.1
+ compile
+
diff --git a/src/main/java/net/tccn/RedisTest.java b/src/main/java/net/tccn/RedisTest.java
index f7e75fb..09894b9 100644
--- a/src/main/java/net/tccn/RedisTest.java
+++ b/src/main/java/net/tccn/RedisTest.java
@@ -1,5 +1,6 @@
package net.tccn;
+import org.junit.Test;
import org.redkale.net.AsyncIOGroup;
import org.redkale.util.AnyValue;
import org.redkale.util.ResourceFactory;
@@ -12,9 +13,65 @@ public class RedisTest {
static MyRedisCacheSource source = new MyRedisCacheSource();
+ /**
+ * 3
+ */
+ @Test
+ public void keyTest() {
+ source.set("a", 3);
+ System.out.println(source.get("a"));
+ source.del("a");
+ }
+
+ /**
+ * ax:false
+ * ax:true
+ * ax:false
+ */
+ @Test
+ public void bitTest() {
+ boolean ax = source.getBit("ax", 6);
+ System.out.println("ax:"+ ax); // false
+ source.setBit("ax", 6, true);
+
+ ax = source.getBit("ax", 6);
+ System.out.println("ax:"+ ax); // true
+
+ source.setBit("ax", 6, false);
+ ax = source.getBit("ax", 6);
+ System.out.println("ax:"+ ax); // false
+ source.del("ax");
+ }
+
+ @Test
+ public void setTest() {
+ source.del("setx");
+ source.sadd("setx", int.class, 1, 2, 3, 5, 6);
+ int setx = source.spop("setx", int.class);
+ System.out.println(setx);
+
+ setx = source.spop("setx", int.class);
+ System.out.println(setx);
+ source.del("setx");
+
+
+ source.srem("setx", int.class,213, 2312);
+
+ /*//source.sadd("setx", list.toArray(Integer[]::new));
+ List list = List.of(2, 3, 5);
+ // source.sadd("setx", list.toArray(Integer[]::new));
+ source.sadd("setx", list.toArray(Integer[]::new));
+ source.sadd("setx", 12, 2312, 213);
+ source.sadd("setx", List.of(1011, 10222));*/
+
+ }
+
+
+
static { // redis://:*Zhong9307!@47.111.150.118:6064?db=2
AnyValue.DefaultAnyValue conf = new AnyValue.DefaultAnyValue().addValue(CACHE_SOURCE_MAXCONNS, "1");
- conf.addValue(CACHE_SOURCE_NODE, new AnyValue.DefaultAnyValue().addValue(CACHE_SOURCE_URL, "redis://:123456@127.0.0.1:6379?db=0"));
+ conf.addValue(CACHE_SOURCE_NODES, "redis://:123456@127.0.0.1:6379?db=0");
+
final ResourceFactory factory = ResourceFactory.create();
final AsyncIOGroup asyncGroup = new AsyncIOGroup(8192, 16);
asyncGroup.start();
@@ -23,22 +80,6 @@ public class RedisTest {
//source.defaultConvert = JsonFactory.root().getConvert();
source.init(conf);
- System.out.println(source.get("a"));
-
- //--------------------- bit ------------------------------
- /*boolean ax = source.getBit("ax", 6);
- System.out.println("ax:"+ ax);
- source.setBit("ax", 6, true);
-
- ax = source.getBit("ax", 6);
- System.out.println("ax:"+ ax);
-
- source.setBit("ax", 6, false);
- ax = source.getBit("ax", 6);
- System.out.println("ax:"+ ax);*/
- //--------------------- bit ------------------------------
-
- //--------------------- bit ------------------------------
/*
source.lock("lockx", 5000);
@@ -52,67 +93,15 @@ public class RedisTest {
int i = (short) 3;
});
- source.keysStartsWith("districtbeans").forEach(x -> {
- System.out.println(x);
- source.del(x);
- });
- source.keysStartsWith("oss-blind-users").forEach(x -> {
- System.out.println(x);
- source.del(x);
- });
- source.keysStartsWith("venue:drama-product-schedule-fee:rt-lbeuai84:20221217").forEach(x -> {
- System.out.println(x);
- source.del(x);
- });
- source.keysStartsWith("venue:site-hour-fee:54f6e9b74cd7416db8d605366c1f49c2:20220729").forEach(x -> {
- System.out.println(x);
- source.del(x);
- });
- source.keysStartsWith("rainbow").forEach(x -> {
- System.out.println(x);
- source.del(x);
- });
- source.keysStartsWith("run").forEach(x -> {
- System.out.println(x);
- source.del(x);
- });
- source.keysStartsWith("today").forEach(x -> {
- System.out.println(x);
- source.del(x);
- });
- source.keysStartsWith("user").forEach(x -> {
- System.out.println(x);
- source.del(x);
- });
- source.keysStartsWith("im:").forEach(x -> {
- System.out.println(x);
- source.del(x);
- });
- source.keysStartsWith("ios-").forEach(x -> {
- System.out.println(x);
- source.del(x);
- });
-
-
//--------------------- set ------------------------------
- /*source.del("setx");
- *//*
- int[] ints = {1, 2, 3};
- source.sadd("setx", ints);
- *//*
+ /*
+ */
+
+ /*
+
- //source.sadd("setx", list.toArray(Integer[]::new));
- List list = List.of(2, 3, 5);
- // source.sadd("setx", list.toArray(Integer[]::new));
- source.sadd("setx", list.toArray(Integer[]::new));
- source.sadd("setx", 12, 2312, 213);
- source.sadd("setx", List.of(1011, 10222));
- source.keys("setx*").forEach(x -> {
- System.out.println(x);
- });
- source.srem("setx", 213, 2312);
Collection setx1 = source.getCollection("setx", String.class);
diff --git a/src/main/java/net/tccn/ZhubAgentProvider.java b/src/main/java/net/tccn/ZhubAgentProvider.java
index 7f9f3c1..5eff32b 100644
--- a/src/main/java/net/tccn/ZhubAgentProvider.java
+++ b/src/main/java/net/tccn/ZhubAgentProvider.java
@@ -29,11 +29,6 @@ public abstract class ZhubAgentProvider extends ClusterAgent {
}
- @Override
- public CompletableFuture>> queryMqtpAddress(String protocol, String module, String resname) {
- return null;
- }
-
@Override
public CompletableFuture> queryHttpAddress(String protocol, String module, String resname) {
return null;
diff --git a/src/main/java/org/redkalex/cache/redis/AbstractRedisSource.java b/src/main/java/org/redkalex/cache/redis/AbstractRedisSource.java
index 53b9398..87c1f1b 100644
--- a/src/main/java/org/redkalex/cache/redis/AbstractRedisSource.java
+++ b/src/main/java/org/redkalex/cache/redis/AbstractRedisSource.java
@@ -8,16 +8,26 @@ import org.redkale.annotation.Resource;
import org.redkale.convert.Convert;
import org.redkale.convert.json.JsonConvert;
import org.redkale.source.AbstractCacheSource;
-import org.redkale.util.AnyValue;
-import org.redkale.util.RedkaleClassLoader;
-import org.redkale.util.RedkaleException;
-import org.redkale.util.ResourceFactory;
+import org.redkale.util.*;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+
+import static org.redkale.boot.Application.RESNAME_APP_EXECUTOR;
+import static org.redkale.boot.Application.RESNAME_APP_NAME;
/**
+ *
* @author zhangjx
+ *
* @since 2.8.0
*/
public abstract class AbstractRedisSource extends AbstractCacheSource {
@@ -26,27 +36,39 @@ public abstract class AbstractRedisSource extends AbstractCacheSource {
protected String name;
+ @Resource(name = RESNAME_APP_NAME, required = false)
+ protected String appName = "";
+
@Resource(required = false)
protected ResourceFactory resourceFactory;
@Resource(required = false)
protected JsonConvert defaultConvert;
- @Resource(name = "$_convert", required = false)
+ @Resource(name = Resource.PARENT_NAME + "_convert", required = false)
protected JsonConvert convert;
protected int db;
protected RedisCryptor cryptor;
- protected AnyValue config;
+ protected AnyValue conf;
+
+ private ExecutorService subExecutor;
+
+ private final ReentrantLock subExecutorLock = new ReentrantLock();
+
+ @Resource(name = RESNAME_APP_EXECUTOR, required = false)
+ protected ExecutorService workExecutor;
@Override
public void init(AnyValue conf) {
- this.config = conf;
+ this.conf = conf;
super.init(conf);
this.name = conf.getValue("name", "");
- if (this.convert == null) this.convert = this.defaultConvert;
+ if (this.convert == null) {
+ this.convert = this.defaultConvert;
+ }
if (conf != null) {
String cryptStr = conf.getValue(CACHE_SOURCE_CRYPTOR, "").trim();
if (!cryptStr.isEmpty()) {
@@ -75,6 +97,46 @@ public abstract class AbstractRedisSource extends AbstractCacheSource {
}
}
+ public boolean acceptsConf(AnyValue config) {
+ if (config == null) {
+ return false;
+ }
+ return "redis".equalsIgnoreCase(config.getValue(CACHE_SOURCE_TYPE))
+ || getClass().getName().equalsIgnoreCase(config.getValue(CACHE_SOURCE_TYPE))
+ || config.getValue(CACHE_SOURCE_NODES, config.getValue("url", "")).startsWith("redis://")
+ || config.getValue(CACHE_SOURCE_NODES, config.getValue("url", "")).startsWith("rediss://");
+ }
+
+ protected ExecutorService subExecutor() {
+ ExecutorService executor = subExecutor;
+ if (executor != null) {
+ return executor;
+ }
+ subExecutorLock.lock();
+ try {
+ if (subExecutor == null) {
+ String threadNameFormat = "CacheSource-" + resourceName() + "-SubThread-%s";
+ Function func = Utility.virtualExecutorFunction();
+ final AtomicInteger counter = new AtomicInteger();
+ subExecutor = func == null ? Executors.newFixedThreadPool(Utility.cpus(), r -> {
+ Thread t = new Thread(r);
+ t.setDaemon(true);
+ int c = counter.incrementAndGet();
+ t.setName(String.format(threadNameFormat, "Virtual-" + (c < 10 ? ("00" + c) : (c < 100 ? ("0" + c) : c))));
+ return t;
+ }) : func.apply(threadNameFormat);
+ }
+ executor = subExecutor;
+ } finally {
+ subExecutorLock.unlock();
+ }
+ return executor;
+ }
+
+ protected String getNodes(AnyValue config) {
+ return config.getValue(CACHE_SOURCE_NODES, config.getValue("url", ""));
+ }
+
@Override
public void close() throws Exception { //在 Application 关闭时调用
destroy(null);
@@ -94,12 +156,22 @@ public abstract class AbstractRedisSource extends AbstractCacheSource {
}
protected T decryptValue(String key, RedisCryptor cryptor, Convert c, Type type, byte[] bs) {
- if (bs == null) return null;
- if (type == byte[].class) return (T) bs;
+ if (bs == null) {
+ return null;
+ }
+ if (type == byte[].class) {
+ return (T) bs;
+ }
+ if (cryptor == null && type == String.class) {
+ return (T) new String(bs, StandardCharsets.UTF_8);
+ }
if (cryptor == null || (type instanceof Class && (((Class) type).isPrimitive() || Number.class.isAssignableFrom((Class) type)))) {
return (T) (c == null ? this.convert : c).convertFrom(type, bs);
}
String deval = cryptor.decrypt(key, new String(bs, StandardCharsets.UTF_8));
+ if (type == String.class) {
+ return (T) deval;
+ }
return deval == null ? null : (T) (c == null ? this.convert : c).convertFrom(type, deval.getBytes(StandardCharsets.UTF_8));
}
@@ -112,20 +184,51 @@ public abstract class AbstractRedisSource extends AbstractCacheSource {
}
protected byte[] encryptValue(String key, RedisCryptor cryptor, Type type, Convert c, T value) {
- if (value == null) return null;
+ if (value == null) {
+ return null;
+ }
Type t = type == null ? value.getClass() : type;
- if (cryptor == null && type == String.class) {
+ if (cryptor == null && t == String.class) {
return value.toString().getBytes(StandardCharsets.UTF_8);
}
- return encryptValue(key, cryptor, t, (c == null ? this.convert : c).convertToBytes(t, value));
+ byte[] bs = (c == null ? this.convert : c).convertToBytes(t, value);
+ if (bs.length > 1 && t instanceof Class && !CharSequence.class.isAssignableFrom((Class) t)) {
+ if (bs[0] == '"' && bs[bs.length - 1] == '"') {
+ bs = Arrays.copyOfRange(bs, 1, bs.length - 1);
+ }
+ }
+ return encryptValue(key, cryptor, t, bs);
}
protected byte[] encryptValue(String key, RedisCryptor cryptor, Type type, byte[] bs) {
- if (bs == null) return null;
+ if (bs == null) {
+ return null;
+ }
if (cryptor == null || (type instanceof Class && (((Class) type).isPrimitive() || Number.class.isAssignableFrom((Class) type)))) {
return bs;
}
String enval = cryptor.encrypt(key, new String(bs, StandardCharsets.UTF_8));
return enval == null ? null : enval.getBytes(StandardCharsets.UTF_8);
}
+
+ protected T decryptScore(Class scoreType, Double score) {
+ if (score == null) {
+ return null;
+ }
+ if (scoreType == int.class || scoreType == Integer.class) {
+ return (T) (Number) score.intValue();
+ } else if (scoreType == long.class || scoreType == Long.class) {
+ return (T) (Number) score.longValue();
+ } else if (scoreType == float.class || scoreType == Float.class) {
+ return (T) (Number) score.floatValue();
+ } else if (scoreType == double.class || scoreType == Double.class) {
+ return (T) (Number) score;
+ } else {
+ return JsonConvert.root().convertFrom(scoreType, score.toString());
+ }
+ }
+
+ protected CompletableFuture returnFutureSize(List> futures) {
+ return futures == null || futures.isEmpty() ? CompletableFuture.completedFuture(0) : Utility.allOfFutures(futures).thenApply(v -> futures.size());
+ }
}
diff --git a/src/main/java/org/redkalex/cache/redis/MyRedisCacheSource.java b/src/main/java/org/redkalex/cache/redis/MyRedisCacheSource.java
index 493c350..a81859e 100644
--- a/src/main/java/org/redkalex/cache/redis/MyRedisCacheSource.java
+++ b/src/main/java/org/redkalex/cache/redis/MyRedisCacheSource.java
@@ -145,6 +145,8 @@ public class MyRedisCacheSource extends RedisCacheSource {
* */
// --------------------
+ /*
+ supper had support
public void zadd(String key, Map kv) {
if (kv == null || kv.isEmpty()) {
return;
@@ -155,16 +157,17 @@ public class MyRedisCacheSource extends RedisCacheSource {
args.add(v);
});
- sendAsync("ZADD", key, args.toArray(Serializable[]::new)).join();
+ sendAsync(RedisCommand.ZADD, key, args.toArray(Serializable[]::new)).join();
}
public double zincr(String key, Serializable number, N n) {
- return sendAsync("ZINCRBY", key, number, n).thenApply(x -> x.getDoubleValue(0d)).join();
+ return sendAsync(RedisCommand.ZINCRBY, key, number, n).thenApply(x -> x.getDoubleValue(0d)).join();
}
- public void zrem(String key, Serializable... vs) {
- sendAsync("ZREM", key, vs).join();
- }
+ @Override
+ public long zrem(String key, String... vs) {
+ return sendAsync(RedisCommand.ZREM, key, keysArgs(key, vs)).thenApply(x -> x.getLongValue(0L)).join();
+ }*/
/*public List zexists(String key, T... fields) {
if (fields == null || fields.length == 0) {
@@ -196,27 +199,27 @@ public class MyRedisCacheSource extends RedisCacheSource {
//--------------------- bit ------------------------------
public boolean getBit(String key, int offset) {
- return sendAsync("GETBIT", key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(offset).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getIntValue(0) > 0).join();
+ return sendAsync(RedisCommand.GETBIT, key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(offset).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getIntValue(0) > 0).join();
}
public void setBit(String key, int offset, boolean bool) {
- sendAsync("SETBIT", key, offset, bool ? 1 : 0).join();
+ sendAsync(RedisCommand.SETBIT, key, keysArgs(key, offset + "", bool ? "1" : "0")).join();
}
//--------------------- bit ------------------------------
//--------------------- lock ------------------------------
// 尝试加锁，成功返回0，否则返回上一锁剩余毫秒值
public long tryLock(String key, int millis) {
- Serializable[] obj = {"" +
+ String[] obj = {"" +
"if (redis.call('EXISTS',KEYS[1]) == 0) then " +
"redis.call('PSETEX',KEYS[1],ARGV[1],1); " +
"return 0; " +
"else " +
"return redis.call('PTTL',KEYS[1]); " +
- "end;", 1, key, millis
+ "end;", 1 + "", key, millis + ""
};
- return sendAsync("EVAL", null, obj).thenApply(v -> v.getIntValue(1)).join();
+ return sendAsync(RedisCommand.EVAL, null, keysArgs(null, obj)).thenApply(v -> v.getIntValue(1)).join();
}
// 加锁
@@ -244,29 +247,33 @@ public class MyRedisCacheSource extends RedisCacheSource {
}
public void set(String key, Serializable value) {
- sendAsync("SET", key, value).join();
+ sendAsync(RedisCommand.SET, key, keysArgs(key, value + "")).join();
}
//--------------------- set ------------------------------
- public void sadd(String key, Collection args) {
- saddAsync(key, args.toArray(Serializable[]::new)).join();
+ /*public void sadd(String key, Collection args) {
+ saddAsync(key, args.toArray(T[]::new)).join();
+ }*/
+
+ /*public void sadd(String key, Serializable... args) {
+ String[] arr = new String[args.length];
+ for (int i = 0; i < args.length; i++) {
+ arr[i] = args[i] + "";
+ }
+ saddAsync(key, arr).join();
}
- public void sadd(String key, Serializable... args) {
- saddAsync(key, Arrays.stream(args).toArray(Serializable[]::new)).join();
- }
-
- public void srem(String key, Serializable... args) {
+ public void srem(String key, String... args) {
sremAsync(key, args).join();
}
public CompletableFuture saddAsync(String key, Serializable... args) {
- return sendAsync("SADD", key, args);
+ return sendAsync(RedisCommand.SADD, key, keysArgs(key, args));
}
- public CompletableFuture sremAsync(String key, Serializable... args) {
- return sendAsync("SREM", key, args);
- }
+ public CompletableFuture sremAsync(String key, String... args) {
+ return sendAsync(RedisCommand.SREM, key, keysArgs(key, args));
+ }*/
//--------------------- hm ------------------------------
@@ -286,14 +293,14 @@ public class MyRedisCacheSource extends RedisCacheSource {
setHmsAsync(key, kv).join();
}
- public CompletableFuture setHmsAsync(String key, Map kv) {
- List args = new ArrayList();
+ public CompletableFuture setHmsAsync(String key, Map kv) {
+ List args = new ArrayList();
kv.forEach((k, v) -> {
args.add(k);
- args.add(v);
+ args.add(v + "");
});
- return sendAsync("HMSET", key, args.toArray(Serializable[]::new));
+ return sendAsync(RedisCommand.HMSET, key, keysArgs(key, args.toArray(String[]::new)));
}
public String getHm(String key, String field) {
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCRC16.java b/src/main/java/org/redkalex/cache/redis/RedisCRC16.java
index 2c0d38f..a0e74db 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCRC16.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCRC16.java
@@ -8,32 +8,35 @@ import org.redkale.util.Utility;
import java.util.Arrays;
/**
+ *
* @author zhangjx
*/
public class RedisCRC16 {
+ private static final int MAX_SLOT = 16384;
+
private static final int[] LOOKUP_TABLE = {0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50A5, 0x60C6,
- 0x70E7, 0x8108, 0x9129, 0xA14A, 0xB16B, 0xC18C, 0xD1AD, 0xE1CE, 0xF1EF, 0x1231, 0x0210, 0x3273,
- 0x2252, 0x52B5, 0x4294, 0x72F7, 0x62D6, 0x9339, 0x8318, 0xB37B, 0xA35A, 0xD3BD, 0xC39C, 0xF3FF,
- 0xE3DE, 0x2462, 0x3443, 0x0420, 0x1401, 0x64E6, 0x74C7, 0x44A4, 0x5485, 0xA56A, 0xB54B, 0x8528,
- 0x9509, 0xE5EE, 0xF5CF, 0xC5AC, 0xD58D, 0x3653, 0x2672, 0x1611, 0x0630, 0x76D7, 0x66F6, 0x5695,
- 0x46B4, 0xB75B, 0xA77A, 0x9719, 0x8738, 0xF7DF, 0xE7FE, 0xD79D, 0xC7BC, 0x48C4, 0x58E5, 0x6886,
- 0x78A7, 0x0840, 0x1861, 0x2802, 0x3823, 0xC9CC, 0xD9ED, 0xE98E, 0xF9AF, 0x8948, 0x9969, 0xA90A,
- 0xB92B, 0x5AF5, 0x4AD4, 0x7AB7, 0x6A96, 0x1A71, 0x0A50, 0x3A33, 0x2A12, 0xDBFD, 0xCBDC, 0xFBBF,
- 0xEB9E, 0x9B79, 0x8B58, 0xBB3B, 0xAB1A, 0x6CA6, 0x7C87, 0x4CE4, 0x5CC5, 0x2C22, 0x3C03, 0x0C60,
- 0x1C41, 0xEDAE, 0xFD8F, 0xCDEC, 0xDDCD, 0xAD2A, 0xBD0B, 0x8D68, 0x9D49, 0x7E97, 0x6EB6, 0x5ED5,
- 0x4EF4, 0x3E13, 0x2E32, 0x1E51, 0x0E70, 0xFF9F, 0xEFBE, 0xDFDD, 0xCFFC, 0xBF1B, 0xAF3A, 0x9F59,
- 0x8F78, 0x9188, 0x81A9, 0xB1CA, 0xA1EB, 0xD10C, 0xC12D, 0xF14E, 0xE16F, 0x1080, 0x00A1, 0x30C2,
- 0x20E3, 0x5004, 0x4025, 0x7046, 0x6067, 0x83B9, 0x9398, 0xA3FB, 0xB3DA, 0xC33D, 0xD31C, 0xE37F,
- 0xF35E, 0x02B1, 0x1290, 0x22F3, 0x32D2, 0x4235, 0x5214, 0x6277, 0x7256, 0xB5EA, 0xA5CB, 0x95A8,
- 0x8589, 0xF56E, 0xE54F, 0xD52C, 0xC50D, 0x34E2, 0x24C3, 0x14A0, 0x0481, 0x7466, 0x6447, 0x5424,
- 0x4405, 0xA7DB, 0xB7FA, 0x8799, 0x97B8, 0xE75F, 0xF77E, 0xC71D, 0xD73C, 0x26D3, 0x36F2, 0x0691,
- 0x16B0, 0x6657, 0x7676, 0x4615, 0x5634, 0xD94C, 0xC96D, 0xF90E, 0xE92F, 0x99C8, 0x89E9, 0xB98A,
- 0xA9AB, 0x5844, 0x4865, 0x7806, 0x6827, 0x18C0, 0x08E1, 0x3882, 0x28A3, 0xCB7D, 0xDB5C, 0xEB3F,
- 0xFB1E, 0x8BF9, 0x9BD8, 0xABBB, 0xBB9A, 0x4A75, 0x5A54, 0x6A37, 0x7A16, 0x0AF1, 0x1AD0, 0x2AB3,
- 0x3A92, 0xFD2E, 0xED0F, 0xDD6C, 0xCD4D, 0xBDAA, 0xAD8B, 0x9DE8, 0x8DC9, 0x7C26, 0x6C07, 0x5C64,
- 0x4C45, 0x3CA2, 0x2C83, 0x1CE0, 0x0CC1, 0xEF1F, 0xFF3E, 0xCF5D, 0xDF7C, 0xAF9B, 0xBFBA, 0x8FD9,
- 0x9FF8, 0x6E17, 0x7E36, 0x4E55, 0x5E74, 0x2E93, 0x3EB2, 0x0ED1, 0x1EF0};
+ 0x70E7, 0x8108, 0x9129, 0xA14A, 0xB16B, 0xC18C, 0xD1AD, 0xE1CE, 0xF1EF, 0x1231, 0x0210, 0x3273,
+ 0x2252, 0x52B5, 0x4294, 0x72F7, 0x62D6, 0x9339, 0x8318, 0xB37B, 0xA35A, 0xD3BD, 0xC39C, 0xF3FF,
+ 0xE3DE, 0x2462, 0x3443, 0x0420, 0x1401, 0x64E6, 0x74C7, 0x44A4, 0x5485, 0xA56A, 0xB54B, 0x8528,
+ 0x9509, 0xE5EE, 0xF5CF, 0xC5AC, 0xD58D, 0x3653, 0x2672, 0x1611, 0x0630, 0x76D7, 0x66F6, 0x5695,
+ 0x46B4, 0xB75B, 0xA77A, 0x9719, 0x8738, 0xF7DF, 0xE7FE, 0xD79D, 0xC7BC, 0x48C4, 0x58E5, 0x6886,
+ 0x78A7, 0x0840, 0x1861, 0x2802, 0x3823, 0xC9CC, 0xD9ED, 0xE98E, 0xF9AF, 0x8948, 0x9969, 0xA90A,
+ 0xB92B, 0x5AF5, 0x4AD4, 0x7AB7, 0x6A96, 0x1A71, 0x0A50, 0x3A33, 0x2A12, 0xDBFD, 0xCBDC, 0xFBBF,
+ 0xEB9E, 0x9B79, 0x8B58, 0xBB3B, 0xAB1A, 0x6CA6, 0x7C87, 0x4CE4, 0x5CC5, 0x2C22, 0x3C03, 0x0C60,
+ 0x1C41, 0xEDAE, 0xFD8F, 0xCDEC, 0xDDCD, 0xAD2A, 0xBD0B, 0x8D68, 0x9D49, 0x7E97, 0x6EB6, 0x5ED5,
+ 0x4EF4, 0x3E13, 0x2E32, 0x1E51, 0x0E70, 0xFF9F, 0xEFBE, 0xDFDD, 0xCFFC, 0xBF1B, 0xAF3A, 0x9F59,
+ 0x8F78, 0x9188, 0x81A9, 0xB1CA, 0xA1EB, 0xD10C, 0xC12D, 0xF14E, 0xE16F, 0x1080, 0x00A1, 0x30C2,
+ 0x20E3, 0x5004, 0x4025, 0x7046, 0x6067, 0x83B9, 0x9398, 0xA3FB, 0xB3DA, 0xC33D, 0xD31C, 0xE37F,
+ 0xF35E, 0x02B1, 0x1290, 0x22F3, 0x32D2, 0x4235, 0x5214, 0x6277, 0x7256, 0xB5EA, 0xA5CB, 0x95A8,
+ 0x8589, 0xF56E, 0xE54F, 0xD52C, 0xC50D, 0x34E2, 0x24C3, 0x14A0, 0x0481, 0x7466, 0x6447, 0x5424,
+ 0x4405, 0xA7DB, 0xB7FA, 0x8799, 0x97B8, 0xE75F, 0xF77E, 0xC71D, 0xD73C, 0x26D3, 0x36F2, 0x0691,
+ 0x16B0, 0x6657, 0x7676, 0x4615, 0x5634, 0xD94C, 0xC96D, 0xF90E, 0xE92F, 0x99C8, 0x89E9, 0xB98A,
+ 0xA9AB, 0x5844, 0x4865, 0x7806, 0x6827, 0x18C0, 0x08E1, 0x3882, 0x28A3, 0xCB7D, 0xDB5C, 0xEB3F,
+ 0xFB1E, 0x8BF9, 0x9BD8, 0xABBB, 0xBB9A, 0x4A75, 0x5A54, 0x6A37, 0x7A16, 0x0AF1, 0x1AD0, 0x2AB3,
+ 0x3A92, 0xFD2E, 0xED0F, 0xDD6C, 0xCD4D, 0xBDAA, 0xAD8B, 0x9DE8, 0x8DC9, 0x7C26, 0x6C07, 0x5C64,
+ 0x4C45, 0x3CA2, 0x2C83, 0x1CE0, 0x0CC1, 0xEF1F, 0xFF3E, 0xCF5D, 0xDF7C, 0xAF9B, 0xBFBA, 0x8FD9,
+ 0x9FF8, 0x6E17, 0x7E36, 0x4E55, 0x5E74, 0x2E93, 0x3EB2, 0x0ED1, 0x1EF0};
private RedisCRC16() {
}
@@ -46,7 +49,7 @@ public class RedisCRC16 {
return crc & 0xFFFF;
}
- public static int calcSlot(int maxSlot, byte[] key) {
+ public static int calcSlot(byte[] key) {
if (key == null) {
return 0;
}
@@ -57,11 +60,10 @@ public class RedisCRC16 {
key = Arrays.copyOfRange(key, start + 1, end);
}
}
- int result = crc16(key) % maxSlot;
- return result;
+ return crc16(key) % MAX_SLOT;
}
- public static int calcSlot(int maxSlot, String key) {
+ public static int calcSlot(String key) {
if (key == null) {
return 0;
}
@@ -72,7 +74,6 @@ public class RedisCRC16 {
key = key.substring(start + 1, end);
}
}
- int result = crc16(key.getBytes()) % maxSlot;
- return result;
+ return crc16(key.getBytes()) % MAX_SLOT;
}
}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheClient.java b/src/main/java/org/redkalex/cache/redis/RedisCacheClient.java
index 3bdca16..d089cf3 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheClient.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheClient.java
@@ -9,23 +9,45 @@ import org.redkale.net.AsyncConnection;
import org.redkale.net.AsyncGroup;
import org.redkale.net.client.Client;
import org.redkale.net.client.ClientAddress;
+import org.redkale.util.Traces;
/**
+ *
* @author zhangjx
*/
public class RedisCacheClient extends Client {
- public RedisCacheClient(String name, AsyncGroup group, String key, ClientAddress address, int maxConns, int maxPipelines, RedisCacheReqAuth authReq, RedisCacheReqDB dbReq) {
+ public RedisCacheClient(String appName, String name, AsyncGroup group, String key,
+ ClientAddress address, int maxConns, int maxPipelines, RedisCacheReqAuth authReq, RedisCacheReqDB dbReq) {
super(name, group, true, address, maxConns, maxPipelines, () -> new RedisCacheReqPing(), () -> new RedisCacheReqClose(), null); //maxConns
- if (authReq != null || dbReq != null) {
- if (authReq != null && dbReq != null) {
- this.authenticate = conn -> writeChannel(conn, authReq).thenCompose(v -> writeChannel(conn, dbReq)).thenApply(v -> conn);
- } else if (authReq != null) {
- this.authenticate = conn -> writeChannel(conn, authReq).thenApply(v -> conn);
- } else {
- this.authenticate = conn -> writeChannel(conn, dbReq).thenApply(v -> conn);
- }
+ RedisCacheReqClientName clientNameReq = new RedisCacheReqClientName(appName, name);
+ if (authReq != null && dbReq != null) {
+ this.authenticate = traceid -> {
+ Traces.currentTraceid(traceid);
+ return conn -> writeChannelBatch(conn, authReq.createTime(), dbReq.createTime(), clientNameReq.createTime())
+ .thenApply(v -> conn);
+ };
+ } else if (authReq != null) {
+ this.authenticate = traceid -> {
+ Traces.currentTraceid(traceid);
+ return conn -> writeChannelBatch(conn, authReq.createTime(), clientNameReq.createTime())
+ .thenApply(v -> conn);
+ };
+ } else if (dbReq != null) {
+ this.authenticate = traceid -> {
+ Traces.currentTraceid(traceid);
+ return conn -> writeChannelBatch(conn, dbReq.createTime(), clientNameReq.createTime())
+ .thenApply(v -> conn);
+ };
+ } else {
+ this.authenticate = traceid -> {
+ Traces.currentTraceid(traceid);
+ return conn -> writeChannel(conn, clientNameReq.createTime())
+ .thenApply(v -> conn);
+ };
}
+ this.readTimeoutSeconds = 3;
+ this.writeTimeoutSeconds = 3;
}
@Override
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheCodec.java b/src/main/java/org/redkalex/cache/redis/RedisCacheCodec.java
index 41902c7..793df77 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheCodec.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheCodec.java
@@ -8,6 +8,7 @@ package org.redkalex.cache.redis;
import org.redkale.net.client.ClientCodec;
import org.redkale.net.client.ClientConnection;
import org.redkale.util.ByteArray;
+import org.redkale.util.RedkaleException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
@@ -16,39 +17,42 @@ import java.util.List;
import java.util.logging.Logger;
/**
+ *
* @author zhangjx
*/
public class RedisCacheCodec extends ClientCodec {
- protected static final byte TYPE_STRING = '+'; //简单字符串(不包含CRLF)类型
+ protected static final byte TYPE_BULK = '$'; //字符串块类型, 例如：$6\r\n\abcdef\r\n，NULL字符串：$-1\r\n
- protected static final byte TYPE_ERROR = '-'; //错误(不包含CRLF)类型
+ protected static final byte TYPE_MULTI = '*'; //数组，紧接的数字为数组长度
- protected static final byte TYPE_NUMBER = ':'; //整型
+ protected static final byte TYPE_STRING = '+'; //字符串值类型，字符串以\r\n结尾, 例如：+OK\r\n
- protected static final byte TYPE_BULK = '$'; //块字符串
+ protected static final byte TYPE_ERROR = '-'; //错误字符串类型，字符串以\r\n结尾, 例如：-ERR unknown command 'red'\r\n
- protected static final byte TYPE_ARRAY = '*'; //数组
+ protected static final byte TYPE_NUMBER = ':'; //整型， 例如：:2\r\n
- protected static final Logger logger = Logger.getLogger(RedisCacheCodec.class.getSimpleName());
+ private static final Logger logger = Logger.getLogger(RedisCacheCodec.class.getSimpleName());
- protected byte halfFrameCmd;
+ private ByteArray halfFrameBytes;
- protected int halfFrameBulkLength = -10;
+ private int halfFrameBulkLength = Integer.MIN_VALUE;
- protected int halfFrameArraySize = -10;
+ private int halfFrameMultiSize = Integer.MIN_VALUE;
- protected int halfFrameArrayIndex; //从0开始
+ private int halfFrameMultiItemIndex; //从0开始
- protected int halfFrameArrayItemLength = -10;
+ private byte halfFrameMultiItemType;
- protected ByteArray halfFrameBytes;
+ private int halfFrameMultiItemLength = Integer.MIN_VALUE;
- protected byte frameType;
+ private byte frameType;
- protected byte[] frameValue; //(不包含CRLF)
+ private byte[] frameCursor;
- protected List frameList; //(不包含CRLF)
+ private byte[] frameValue;
+
+ private List frameList;
private ByteArray recyclableArray;
@@ -56,148 +60,142 @@ public class RedisCacheCodec extends ClientCodec deque = (ArrayDeque) responseQueue(conn);
-// logger.log(Level.FINEST, "[" + Utility.nowMillis() + "] [" + Thread.currentThread().getName() + "]: " + conn + ", 原始数据: " + new String(dbs).replace("\r\n", " ") + ", req=" + deque.getFirst().getRequest());
+// (System. out).println("[" + Utility.nowMillis() + "] [" + Thread.currentThread().getName() + "]: " + conn + ", 原始数据: " + new String(dbs).replace("\r\n", " "));
array.clear();
- byte type = halfFrameCmd == 0 ? buffer.get() : halfFrameCmd;
- if (halfFrameBytes != null) {
- array.put(halfFrameBytes, 0, halfFrameBytes.length());
+ if (this.frameType == 0) {
+ this.frameType = buffer.get();
+ } else if (halfFrameBytes != null) {
+ array.put(halfFrameBytes);
+ halfFrameBytes = null;
}
- frameType = type;
- if (type == TYPE_STRING || type == TYPE_ERROR || type == TYPE_NUMBER) {
- if (readComplete(buffer, array)) {
- frameValue = array.getBytes();
- } else {
- halfFrameCmd = type;
+ if (frameType == TYPE_STRING || frameType == TYPE_ERROR || frameType == TYPE_NUMBER) {
+ if (!readComplete(buffer, array)) {
halfFrameBytes = pollArray(array);
return false;
}
- } else if (type == TYPE_BULK) {
- int bulkLength = halfFrameBulkLength;
- if (bulkLength < -2) {
+ frameValue = array.getBytes();
+ } else if (frameType == TYPE_BULK) {
+ if (halfFrameBulkLength == Integer.MIN_VALUE) {
if (!readComplete(buffer, array)) { //没有读到bulkLength
- halfFrameCmd = type;
- halfFrameBulkLength = -10;
halfFrameBytes = pollArray(array);
return false;
}
- bulkLength = Integer.parseInt(array.toString(StandardCharsets.UTF_8));
+ halfFrameBulkLength = readInt(array);
array.clear();
}
- if (bulkLength == -1) {
+ if (halfFrameBulkLength == -1) {
frameValue = null;
- } else if (readComplete(buffer, array)) {
- frameValue = array.getBytes();
} else {
- halfFrameCmd = type;
- halfFrameBulkLength = bulkLength;
- halfFrameBytes = pollArray(array);
- return false;
- }
- } else if (type == TYPE_ARRAY) {
- int arraySize = halfFrameArraySize;
- if (arraySize < -2) {
- if (!readComplete(buffer, array)) { //没有读到arraySize
- halfFrameCmd = type;
- halfFrameArraySize = -10;
- halfFrameArrayIndex = 0;
- halfFrameArrayItemLength = -10;
+ int expect = halfFrameBulkLength + 2 - array.length();
+ if (buffer.remaining() < expect) {
+ array.put(buffer);
halfFrameBytes = pollArray(array);
return false;
}
- arraySize = Integer.parseInt(array.toString(StandardCharsets.UTF_8));
- array.clear();
+ array.put(buffer, expect);
+ array.removeLastByte(); //移除\n
+ array.removeLastByte(); //移除\r
+ frameValue = array.getBytes();
}
- int arrayIndex = halfFrameArrayIndex;
- for (int i = arrayIndex; i < arraySize; i++) {
- int itemLength = halfFrameArrayItemLength;
- halfFrameArrayItemLength = -10;
- if (itemLength < -2) {
- if (!readComplete(buffer, array)) { //没有读到bulkLength
- halfFrameCmd = type;
- halfFrameArraySize = arraySize;
- halfFrameArrayIndex = i;
- halfFrameArrayItemLength = -10;
- halfFrameBytes = pollArray(array);
+ } else if (frameType == TYPE_MULTI) {
+ int size = halfFrameMultiSize;
+ if (size == Integer.MIN_VALUE) {
+ if (!readComplete(buffer, array)) { //没有读到bulkLength
+ halfFrameBytes = pollArray(array);
+ return false;
+ }
+ size = readInt(array);
+ halfFrameMultiSize = size;
+ array.clear();
+ frameValue = null;
+ }
+ if (frameList == null) {
+ frameList = new ArrayList<>();
+ }
+ if (size > 0) {
+ int index = halfFrameMultiItemIndex;
+ for (int i = index; i < size; i++) {
+ if (!buffer.hasRemaining()) {
return false;
}
- byte sign = array.get(0);
- itemLength = Integer.parseInt(array.toString(1, StandardCharsets.UTF_8));
- array.clear();
- if (sign == TYPE_ARRAY) { //数组中嵌套数组，目前有 HSCAN
- frameValue = null;
- if (frameList != null) {
- frameList.clear();
- }
- clearHalfFrame();
- if (itemLength == 0) {
- return true;
- }
- halfFrameCmd = sign;
- halfFrameArraySize = itemLength;
- if (!buffer.hasRemaining()) {
+ if (halfFrameMultiItemType == 0) {
+ halfFrameMultiItemType = buffer.get();
+ }
+ halfFrameMultiItemIndex = i;
+ final byte itemType = halfFrameMultiItemType;
+ if (itemType == TYPE_STRING || itemType == TYPE_ERROR || itemType == TYPE_NUMBER) {
+ if (!readComplete(buffer, array)) {
+ halfFrameBytes = pollArray(array);
return false;
}
- return checkBytesFrame(conn, buffer, array);
+ frameList.add(array.getBytes());
+ } else if (itemType == TYPE_BULK) {
+ if (halfFrameMultiItemLength == Integer.MIN_VALUE) {
+ if (!readComplete(buffer, array)) { //没有读到bulkLength
+ halfFrameBytes = pollArray(array);
+ return false;
+ }
+ halfFrameMultiItemLength = readInt(array);
+ array.clear();
+ }
+ if (halfFrameMultiItemLength == -1) {
+ frameList.add(null);
+ } else {
+ int expect = halfFrameMultiItemLength + 2 - array.length();
+ if (buffer.remaining() < expect) {
+ array.put(buffer);
+ halfFrameBytes = pollArray(array);
+ return false;
+ }
+ array.put(buffer, expect);
+ array.removeLastByte(); //移除\n
+ array.removeLastByte(); //移除\r
+ frameList.add(array.getBytes());
+ }
+ } else if (itemType == TYPE_MULTI) { //数组中嵌套数组，例如: SCAN、HSCAN
+ if (size == 2 && frameList != null && frameList.size() == 1) {
+ //读游标 数据例如: *2 $1 0 *4 $4 key1 $2 10 $4 key2 $2 30
+ frameCursor = frameList.get(0);
+ frameList.clear();
+ clearHalfFrame();
+ return readFrames(conn, buffer, array);
+ } else {
+ throw new RedkaleException("Not support multi type in array data");
+ }
}
- }
- int cha = itemLength - array.length();
- if (itemLength == -1) {
- if (frameList == null) {
- frameList = new ArrayList<>();
- }
- frameList.add(null);
+ halfFrameMultiItemType = 0;
+ halfFrameMultiItemLength = Integer.MIN_VALUE;
array.clear();
- } else if (buffer.remaining() >= cha + 2) {
- for (int j = 0; j < cha; j++) array.put(buffer.get());
- buffer.get(); //\r
- buffer.get(); //\n
- if (frameList == null) {
- frameList = new ArrayList<>();
- }
- frameList.add(array.getBytes());
- array.clear();
- } else {
- while (buffer.hasRemaining()) array.put(buffer.get());
- halfFrameCmd = type;
- halfFrameArraySize = arraySize;
- halfFrameArrayIndex = i;
- halfFrameArrayItemLength = itemLength;
- halfFrameBytes = pollArray(array);
- return false;
}
}
}
- clearHalfFrame();
return true;
}
- protected void clearHalfFrame() {
- halfFrameCmd = 0;
- halfFrameBulkLength = -10;
- halfFrameArraySize = -10;
- halfFrameArrayIndex = 0;
- halfFrameArrayItemLength = -10;
+ private void clearHalfFrame() {
halfFrameBytes = null;
+ halfFrameBulkLength = Integer.MIN_VALUE;
+ halfFrameMultiSize = Integer.MIN_VALUE;
+ halfFrameMultiItemLength = Integer.MIN_VALUE;
+ halfFrameMultiItemIndex = 0; //从0开始
+ halfFrameMultiItemType = 0;
}
@Override
@@ -207,35 +205,31 @@ public class RedisCacheCodec extends ClientCodec {
@@ -32,18 +36,29 @@ public class RedisCacheConnection extends ClientConnection> writeRequest(RedisCacheRequest[] requests) {
+ return super.writeChannel(requests);
+ }
+
protected CompletableFuture writeRequest(RedisCacheRequest request, Function respTransfer) {
return super.writeChannel(request, respTransfer);
}
+ protected CompletableFuture> writeRequest(RedisCacheRequest[] requests, Function respTransfer) {
+ return super.writeChannel(requests, respTransfer);
+ }
+
public RedisCacheResult pollResultSet(RedisCacheRequest request) {
RedisCacheResult rs = new RedisCacheResult();
return rs;
}
- public RedisCacheRequest pollRequest(WorkThread workThread) {
- RedisCacheRequest rs = new RedisCacheRequest().currThread(workThread);
+ public RedisCacheRequest pollRequest(WorkThread workThread, String traceid) {
+ RedisCacheRequest rs = new RedisCacheRequest().workThread(workThread).traceid(traceid);
return rs;
}
+ protected ClientFuture pollRespFuture(Serializable requestid) {
+ return super.pollRespFuture(requestid);
+ }
}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheReqAuth.java b/src/main/java/org/redkalex/cache/redis/RedisCacheReqAuth.java
index 41dbd2f..b83f70d 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheReqAuth.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheReqAuth.java
@@ -11,11 +11,12 @@ import org.redkale.util.ByteArray;
import java.nio.charset.StandardCharsets;
/**
+ *
* @author zhangjx
*/
public class RedisCacheReqAuth extends RedisCacheRequest {
- private static final byte[] PS = "AUTH".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] PS = "AUTH\r\n".getBytes(StandardCharsets.UTF_8);
protected String password;
@@ -26,20 +27,18 @@ public class RedisCacheReqAuth extends RedisCacheRequest {
@Override
public void writeTo(ClientConnection conn, ByteArray writer) {
byte[] pwd = password.getBytes();
- writer.put((byte) '*');
- writer.put((byte) '2');
- writer.put((byte) '\r', (byte) '\n');
- writer.put((byte) '$');
- writer.put((byte) '4');
- writer.put((byte) '\r', (byte) '\n');
+ writer.put(mutliLengthBytes(2));
+ writer.put(bulkLengthBytes(4));
writer.put(PS);
- writer.put((byte) '\r', (byte) '\n');
- writer.put((byte) '$');
- writer.put(String.valueOf(pwd.length).getBytes(StandardCharsets.UTF_8));
- writer.put((byte) '\r', (byte) '\n');
+ writer.put(bulkLengthBytes(pwd.length));
writer.put(pwd);
- writer.put((byte) '\r', (byte) '\n');
+ writer.put(CRLF);
}
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() + "{AUTH " + password + "}";
+ }
}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheReqClientName.java b/src/main/java/org/redkalex/cache/redis/RedisCacheReqClientName.java
new file mode 100644
index 0000000..1407344
--- /dev/null
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheReqClientName.java
@@ -0,0 +1,45 @@
+/*
+ *
+ */
+package org.redkalex.cache.redis;
+
+import org.redkale.net.client.ClientConnection;
+import org.redkale.util.ByteArray;
+import org.redkale.util.Utility;
+
+import java.nio.charset.StandardCharsets;
+
+/**
+ *
+ * @author zhangjx
+ */
+public class RedisCacheReqClientName extends RedisCacheRequest {
+
+ private final String clientName;
+
+ public RedisCacheReqClientName(String appName, String resourceName) {
+ this.clientName = "redkalex" + (Utility.isEmpty(appName) ? "" : ("-" + appName))
+ + (Utility.isEmpty(resourceName) ? "" : (":" + resourceName));
+ }
+
+ @Override
+ public void writeTo(ClientConnection conn, ByteArray writer) {
+ writer.put(mutliLengthBytes(3));
+
+ writer.put(bulkLengthBytes(6));
+ writer.put("CLIENT\r\n".getBytes(StandardCharsets.UTF_8));
+
+ writer.put(bulkLengthBytes(7));
+ writer.put("SETNAME\r\n".getBytes(StandardCharsets.UTF_8));
+
+ byte[] ns = clientName.getBytes(StandardCharsets.UTF_8);
+ writer.put(bulkLengthBytes(ns.length));
+ writer.put(ns);
+ writer.put(CRLF);
+ }
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() + "{CLIENT SETNAME " + clientName + "}";
+ }
+}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheReqClose.java b/src/main/java/org/redkalex/cache/redis/RedisCacheReqClose.java
index 67432f8..f1a304d 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheReqClose.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheReqClose.java
@@ -11,11 +11,20 @@ import org.redkale.util.ByteArray;
import java.nio.charset.StandardCharsets;
/**
+ *
* @author zhangjx
*/
public class RedisCacheReqClose extends RedisCacheRequest {
- private static final byte[] PS = "QUIT".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] BYTES = new ByteArray()
+ .put((byte) '*')
+ .put((byte) '1')
+ .put((byte) '\r', (byte) '\n')
+ .put((byte) '$')
+ .put((byte) '4')
+ .put((byte) '\r', (byte) '\n')
+ .put("QUIT".getBytes(StandardCharsets.UTF_8))
+ .put((byte) '\r', (byte) '\n').getBytes();
@Override
public final boolean isCloseType() {
@@ -24,13 +33,11 @@ public class RedisCacheReqClose extends RedisCacheRequest {
@Override
public void writeTo(ClientConnection conn, ByteArray writer) {
- writer.put((byte) '*');
- writer.put((byte) '1');
- writer.put((byte) '\r', (byte) '\n');
- writer.put((byte) '$');
- writer.put((byte) '4');
- writer.put((byte) '\r', (byte) '\n');
- writer.put(PS);
- writer.put((byte) '\r', (byte) '\n');
+ writer.put(BYTES);
+ }
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() + "{QUIT}";
}
}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheReqDB.java b/src/main/java/org/redkalex/cache/redis/RedisCacheReqDB.java
index 37ef12a..d9848e0 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheReqDB.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheReqDB.java
@@ -11,6 +11,7 @@ import org.redkale.util.ByteArray;
import java.nio.charset.StandardCharsets;
/**
+ *
* @author zhangjx
*/
public class RedisCacheReqDB extends RedisCacheRequest {
@@ -23,21 +24,19 @@ public class RedisCacheReqDB extends RedisCacheRequest {
@Override
public void writeTo(ClientConnection conn, ByteArray writer) {
- writer.put((byte) '*');
- writer.put((byte) '2');
- writer.put((byte) '\r', (byte) '\n');
- writer.put((byte) '$');
- writer.put((byte) '6');
- writer.put((byte) '\r', (byte) '\n');
- writer.put("SELECT".getBytes(StandardCharsets.UTF_8));
- writer.put((byte) '\r', (byte) '\n');
+ writer.put(mutliLengthBytes(2));
+ writer.put(bulkLengthBytes(6));
+ writer.put("SELECT\r\n".getBytes(StandardCharsets.UTF_8));
byte[] dbs = String.valueOf(db).getBytes(StandardCharsets.UTF_8);
- writer.put((byte) '$');
- writer.put(String.valueOf(dbs.length).getBytes(StandardCharsets.UTF_8));
- writer.put((byte) '\r', (byte) '\n');
+ writer.put(bulkLengthBytes(dbs.length));
writer.put(dbs);
- writer.put((byte) '\r', (byte) '\n');
+ writer.put(CRLF);
}
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() + "{SELECT " + db + "}";
+ }
}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheReqPing.java b/src/main/java/org/redkalex/cache/redis/RedisCacheReqPing.java
index 114bb10..b6d469a 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheReqPing.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheReqPing.java
@@ -11,21 +11,28 @@ import org.redkale.util.ByteArray;
import java.nio.charset.StandardCharsets;
/**
+ *
* @author zhangjx
*/
public class RedisCacheReqPing extends RedisCacheRequest {
- private static final byte[] PS = "PING".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] BYTES = new ByteArray()
+ .put((byte) '*')
+ .put((byte) '1')
+ .put((byte) '\r', (byte) '\n')
+ .put((byte) '$')
+ .put((byte) '4')
+ .put((byte) '\r', (byte) '\n')
+ .put("PING".getBytes(StandardCharsets.UTF_8))
+ .put((byte) '\r', (byte) '\n').getBytes();
@Override
public void writeTo(ClientConnection conn, ByteArray writer) {
- writer.put((byte) '*');
- writer.put((byte) '1');
- writer.put((byte) '\r', (byte) '\n');
- writer.put((byte) '$');
- writer.put((byte) '4');
- writer.put((byte) '\r', (byte) '\n');
- writer.put(PS);
- writer.put((byte) '\r', (byte) '\n');
+ writer.put(BYTES);
+ }
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() + "{PING}";
}
}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheRequest.java b/src/main/java/org/redkalex/cache/redis/RedisCacheRequest.java
index 403b7a8..c32029a 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheRequest.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheRequest.java
@@ -12,21 +12,49 @@ import org.redkale.util.ByteArray;
import java.nio.charset.StandardCharsets;
/**
+ *
* @author zhangjx
*/
public class RedisCacheRequest extends ClientRequest {
- static final byte[] TRUE = new byte[]{'t'};
+ static final byte[] BYTES_TRUE = new byte[]{'t'};
- static final byte[] FALSE = new byte[]{'f'};
+ static final byte[] BYTES_FALSE = new byte[]{'f'};
+
+ static final byte[] BYTES_MATCH = "MATCH".getBytes(StandardCharsets.UTF_8);
+
+ static final byte[] BYTES_COUNT = "COUNT".getBytes(StandardCharsets.UTF_8);
+
+ protected static final byte[] CRLF = new byte[]{'\r', '\n'};
+
+ private static final byte[][] starLengthBytes;
+
+ private static final byte[][] dollarLengthBytes;
+
+ static {
+ starLengthBytes = new byte[1024][];
+ dollarLengthBytes = new byte[1024][];
+ for (int i = 0; i < dollarLengthBytes.length; i++) {
+ starLengthBytes[i] = ("*" + i + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
+ dollarLengthBytes[i] = ("$" + i + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
+ }
+ }
+
+ protected RedisCommand command;
protected String key;
- protected String command;
-
protected byte[][] args;
- public RedisCacheRequest prepare(String command, String key, byte[]... args) {
+ public static RedisCacheRequest create(RedisCommand command, String key, String... args) {
+ return new RedisCacheRequest().prepare(command, key, RedisCacheSource.keysArgs(key, args));
+ }
+
+ public static RedisCacheRequest create(RedisCommand command, String key, byte[]... args) {
+ return new RedisCacheRequest().prepare(command, key, args);
+ }
+
+ public RedisCacheRequest prepare(RedisCommand command, String key, byte[]... args) {
super.prepare();
this.command = command;
this.key = key;
@@ -34,28 +62,55 @@ public class RedisCacheRequest extends ClientRequest {
return this;
}
+ public RedisCacheRequest createTime() {
+ this.createTime = System.currentTimeMillis();
+ return this;
+ }
+
@Override
public void writeTo(ClientConnection conn, ByteArray writer) {
- writer.put((byte) '*');
- writer.put(String.valueOf(args.length + 1).getBytes(StandardCharsets.UTF_8));
- writer.put((byte) '\r', (byte) '\n');
- writer.put((byte) '$');
- writer.put(String.valueOf(command.length()).getBytes(StandardCharsets.UTF_8));
- writer.put((byte) '\r', (byte) '\n');
- writer.put(command.getBytes(StandardCharsets.UTF_8));
- writer.put((byte) '\r', (byte) '\n');
+ writer.put(mutliLengthBytes(args.length + 1));
+ writer.put(command.getBytes());
for (final byte[] arg : args) {
- writer.put((byte) '$');
- writer.put(String.valueOf(arg.length).getBytes(StandardCharsets.UTF_8));
- writer.put((byte) '\r', (byte) '\n');
- writer.put(arg);
- writer.put((byte) '\r', (byte) '\n');
+ putArgBytes(writer, arg);
+ }
+ }
+
+ protected void putArgBytes(ByteArray writer, byte[] arg) {
+ writer.put(bulkLengthBytes(arg.length));
+ writer.put(arg);
+ writer.put(CRLF);
+ }
+
+ protected static byte[] mutliLengthBytes(int length) {
+ if (length >= 0 && length < starLengthBytes.length) {
+ return starLengthBytes[length];
+ } else {
+ return ("*" + length + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
+ }
+ }
+
+ protected static byte[] bulkLengthBytes(int length) {
+ if (length >= 0 && length < dollarLengthBytes.length) {
+ return dollarLengthBytes[length];
+ } else {
+ return ("$" + length + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
}
}
@Override
public String toString() {
- return getClass().getSimpleName() + "{command=" + command + ", key=" + key + "}";
+ if (args == null || args.length == 0) {
+ return getClass().getSimpleName() + "{" + command + " " + key + "}";
+ } else {
+ StringBuilder sb = new StringBuilder();
+ sb.append(command);
+ sb.append(" ").append(key);
+ for (final byte[] arg : args) {
+ sb.append(" ").append(arg == null ? null : new String(arg, StandardCharsets.UTF_8));
+ }
+ return sb.toString();
+ }
}
}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheResult.java b/src/main/java/org/redkalex/cache/redis/RedisCacheResult.java
index 96e8445..330569a 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheResult.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheResult.java
@@ -6,34 +6,47 @@
package org.redkalex.cache.redis;
import org.redkale.convert.json.JsonConvert;
+import org.redkale.net.client.ClientResult;
+import org.redkale.source.CacheScoredValue;
import java.lang.reflect.Type;
+import java.math.BigDecimal;
+import java.math.BigInteger;
import java.nio.charset.StandardCharsets;
import java.util.*;
/**
+ *
* @author zhangjx
*/
-public class RedisCacheResult {
+public class RedisCacheResult implements ClientResult {
- //+ 简单字符串类型 (不包含CRLF)
- //- 错误类型 (不包含CRLF)
- //': 整型
- //$ 块字符串
+ //$ 块字符串类型
//* 数组
+ //+ 简单字符串类型
+ //- 错误类型
+ //: 整型
protected byte frameType;
- protected byte[] frameValue; //(不包含CRLF)
+ protected byte[] frameCursor;
- protected List frameList; //(不包含CRLF)
+ protected byte[] frameValue;
- public RedisCacheResult prepare(byte byteType, byte[] val, List bytesList) {
+ protected List frameList;
+
+ public RedisCacheResult prepare(byte byteType, byte[] frameCursor, byte[] frameValue, List frameList) {
this.frameType = byteType;
- this.frameValue = val;
- this.frameList = bytesList;
+ this.frameCursor = frameCursor;
+ this.frameValue = frameValue;
+ this.frameList = frameList;
return this;
}
+ @Override
+ public boolean isKeepAlive() {
+ return true;
+ }
+
public Void getVoidValue() {
return null;
}
@@ -42,6 +55,14 @@ public class RedisCacheResult {
return frameValue;
}
+ public int getCursor() {
+ if (frameCursor == null || frameCursor.length < 1) {
+ return -1;
+ } else {
+ return Integer.parseInt(new String(frameCursor));
+ }
+ }
+
public Boolean getBoolValue() {
if (frameValue == null) {
return false;
@@ -50,34 +71,73 @@ public class RedisCacheResult {
if ("OK".equals(val)) {
return true;
}
+ if (val.isEmpty()) {
+ return false;
+ }
+ for (char ch : val.toCharArray()) {
+ if (!Character.isDigit(ch)) {
+ return false;
+ }
+ }
return Integer.parseInt(val) > 0;
}
- public String getStringValue(String key, RedisCryptor cryptor) {
+ public Double getDoubleValue(Double defValue) {
if (frameValue == null) {
- return null;
+ return defValue;
}
String val = new String(frameValue, StandardCharsets.UTF_8);
- if (cryptor != null) {
- val = cryptor.decrypt(key, val);
+ if ("nan".equalsIgnoreCase(val) || "-nan".equalsIgnoreCase(val)) {
+ return Double.NaN;
+ } else if ("inf".equalsIgnoreCase(val)) {
+ return Double.POSITIVE_INFINITY;
+ } else if ("-inf".equalsIgnoreCase(val)) {
+ return Double.NEGATIVE_INFINITY;
+ } else if ("-1".equalsIgnoreCase(val)) {
+ return -1.0;
+ } else if ("0".equalsIgnoreCase(val)) {
+ return 0.0;
+ } else if ("1".equalsIgnoreCase(val)) {
+ return 1.0;
+ } else {
+ return Double.parseDouble(val);
}
- return val;
}
- public Double getDoubleValue(Double defvalue) {
- return frameValue == null ? defvalue : Double.parseDouble(new String(frameValue, StandardCharsets.UTF_8));
+ public Long getLongValue(Long defValue) {
+ if (frameValue == null) {
+ return defValue;
+ }
+ String val = new String(frameValue, StandardCharsets.UTF_8);
+ if ("-1".equalsIgnoreCase(val)) {
+ return -1L;
+ } else if ("0".equalsIgnoreCase(val)) {
+ return 0L;
+ } else if ("1".equalsIgnoreCase(val)) {
+ return 1L;
+ } else {
+ return Long.parseLong(val);
+ }
}
- public Long getLongValue(Long defvalue) {
- return frameValue == null ? defvalue : Long.parseLong(new String(frameValue, StandardCharsets.UTF_8));
- }
-
- public Integer getIntValue(Integer defvalue) {
- return frameValue == null ? defvalue : Integer.parseInt(new String(frameValue, StandardCharsets.UTF_8));
+ public Integer getIntValue(Integer defValue) {
+ if (frameValue == null) {
+ return defValue;
+ }
+ String val = new String(frameValue, StandardCharsets.UTF_8);
+ if ("-1".equalsIgnoreCase(val)) {
+ return -1;
+ } else if ("0".equalsIgnoreCase(val)) {
+ return 0;
+ } else if ("1".equalsIgnoreCase(val)) {
+ return 1;
+ } else {
+ return Integer.parseInt(val);
+ }
}
public T getObjectValue(String key, RedisCryptor cryptor, Type type) {
- return formatValue(key, cryptor, frameValue, type);
+ return decodeValue(key, cryptor, frameValue, type);
}
protected Set getSetValue(String key, RedisCryptor cryptor, Type type) {
@@ -86,7 +146,23 @@ public class RedisCacheResult {
}
Set set = new LinkedHashSet<>();
for (byte[] bs : frameList) {
- set.add(formatValue(key, cryptor, bs, type));
+ set.add(decodeValue(key, cryptor, bs, type));
+ }
+ return set;
+ }
+
+ protected List getScoreListValue(String key, RedisCryptor cryptor, Type scoreType) {
+ if (frameList == null || frameList.isEmpty()) {
+ return new ArrayList<>();
+ }
+ List set = new ArrayList<>();
+ for (int i = 0; i < frameList.size(); i += 2) {
+ byte[] bs1 = frameList.get(i);
+ byte[] bs2 = frameList.get(i + 1);
+ Number val = decodeValue(key, cryptor, bs2, scoreType);
+ if (val != null) {
+ set.add(CacheScoredValue.create(val, new String(bs1, StandardCharsets.UTF_8)));
+ }
}
return set;
}
@@ -97,7 +173,7 @@ public class RedisCacheResult {
}
List list = new ArrayList<>();
for (byte[] bs : frameList) {
- list.add(formatValue(key, cryptor, bs, type));
+ list.add(decodeValue(key, cryptor, bs, type));
}
return list;
}
@@ -110,15 +186,15 @@ public class RedisCacheResult {
for (int i = 0; i < frameList.size(); i += 2) {
byte[] bs1 = frameList.get(i);
byte[] bs2 = frameList.get(i + 1);
- T val = formatValue(key, cryptor, bs2, type);
+ T val = decodeValue(key, cryptor, bs2, type);
if (val != null) {
- map.put(formatValue(key, cryptor, bs1, String.class).toString(), val);
+ map.put(decodeValue(key, cryptor, bs1, String.class).toString(), val);
}
}
return map;
}
- protected static T formatValue(String key, RedisCryptor cryptor, byte[] frames, Type type) {
+ protected static T decodeValue(String key, RedisCryptor cryptor, byte[] frames, Type type) {
if (frames == null) {
return null;
}
@@ -132,12 +208,25 @@ public class RedisCacheResult {
}
return (T) val;
}
- if (type == boolean.class || type == Boolean.class) {
- return (T) (Boolean) "t".equalsIgnoreCase(new String(frames, StandardCharsets.UTF_8));
+ if (type == int.class || type == Integer.class) {
+ return (T) (Integer) Integer.parseInt(new String(frames, StandardCharsets.UTF_8));
}
if (type == long.class || type == Long.class) {
return (T) (Long) Long.parseLong(new String(frames, StandardCharsets.UTF_8));
}
+ if (type == float.class || type == Float.class) {
+ return (T) (Float) Float.parseFloat(new String(frames, StandardCharsets.UTF_8));
+ }
+ if (type == BigInteger.class) {
+ return (T) new BigInteger(new String(frames, StandardCharsets.UTF_8));
+ }
+ if (type == BigDecimal.class) {
+ return (T) new BigDecimal(new String(frames, StandardCharsets.UTF_8));
+ }
+ if (type == boolean.class || type == Boolean.class) {
+ String v = new String(frames, StandardCharsets.UTF_8);
+ return (T) (Boolean) ("t".equalsIgnoreCase(v) || "1".equals(v));
+ }
if (type == double.class || type == Double.class) {
return (T) (Double) Double.parseDouble(new String(frames, StandardCharsets.UTF_8));
}
@@ -148,4 +237,26 @@ public class RedisCacheResult {
return (T) JsonConvert.root().convertFrom(type, frames);
}
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("{type: ").append(frameType);
+ if (frameValue != null) {
+ sb.append(", value: ").append(new String(frameValue, StandardCharsets.UTF_8));
+ }
+ if (frameList != null) {
+ sb.append(", list: [");
+ boolean first = true;
+ for (byte[] bs : frameList) {
+ if (!first) {
+ sb.append(", ");
+ }
+ sb.append(bs == null ? null : new String(bs, StandardCharsets.UTF_8));
+ first = false;
+ }
+ sb.append("]");
+ }
+ return sb.append("}").toString();
+ }
+
}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheSource.java b/src/main/java/org/redkalex/cache/redis/RedisCacheSource.java
index 0e7211c..68ffeb5 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheSource.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheSource.java
@@ -6,15 +6,17 @@
package org.redkalex.cache.redis;
import org.redkale.annotation.AutoLoad;
-import org.redkale.annotation.Resource;
import org.redkale.annotation.ResourceListener;
import org.redkale.annotation.ResourceType;
+import org.redkale.annotation.*;
import org.redkale.convert.Convert;
import org.redkale.convert.json.JsonConvert;
import org.redkale.net.AsyncGroup;
import org.redkale.net.WorkThread;
-import org.redkale.net.client.ClientAddress;
+import org.redkale.net.client.*;
import org.redkale.service.Local;
+import org.redkale.source.CacheEventListener;
+import org.redkale.source.CacheScoredValue;
import org.redkale.source.CacheSource;
import org.redkale.util.*;
@@ -24,20 +26,24 @@ import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.*;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.Function;
import java.util.logging.Level;
import java.util.logging.Logger;
import static org.redkale.boot.Application.RESNAME_APP_CLIENT_ASYNCGROUP;
import static org.redkale.boot.Application.RESNAME_APP_EXECUTOR;
+import static org.redkale.util.Utility.isEmpty;
+import static org.redkale.util.Utility.isNotEmpty;
+import static org.redkalex.cache.redis.RedisCacheRequest.BYTES_COUNT;
+import static org.redkalex.cache.redis.RedisCacheRequest.BYTES_MATCH;
/**
* 详情见: https://redkale.org
*
+ *
* @author zhangjx
*/
@Local
@@ -47,32 +53,31 @@ public class RedisCacheSource extends AbstractRedisSource {
static final boolean debug = false; //System.getProperty("os.name").contains("Window") || System.getProperty("os.name").contains("Mac");
- protected static final byte FRAME_TYPE_BULK = '$'; //块字符串
-
- protected static final byte FRAME_TYPE_ARRAY = '*'; //数组
-
- protected static final byte FRAME_TYPE_STRING = '+'; //简单字符串(不包含CRLF)类型
-
- protected static final byte FRAME_TYPE_ERROR = '-'; //错误(不包含CRLF)类型
-
- protected static final byte FRAME_TYPE_NUMBER = ':'; //整型
-
protected static final byte[] NX = "NX".getBytes();
protected static final byte[] EX = "EX".getBytes();
- private final Logger logger = Logger.getLogger(this.getClass().getSimpleName());
+ protected static final byte[] CHANNELS = "CHANNELS".getBytes();
+
+ private final Logger logger = Logger.getLogger(getClass().getSimpleName());
@Resource(name = RESNAME_APP_CLIENT_ASYNCGROUP, required = false)
- protected AsyncGroup clientAsyncGroup;
+ private AsyncGroup clientAsyncGroup;
//配置 APP_EXECUTOR资源为null
@Resource(name = RESNAME_APP_EXECUTOR, required = false)
- protected ExecutorService workExecutor;
+ private ExecutorService workExecutor;
- protected RedisCacheClient client;
+ private RedisCacheClient client;
- protected InetSocketAddress address;
+ private InetSocketAddress address;
+
+ private RedisCacheConnection subConn;
+
+ private final ReentrantLock pubsubLock = new ReentrantLock();
+
+ //key: topic
+ private final Map>> pubsubListeners = new ConcurrentHashMap<>();
@Override
public void init(AnyValue conf) {
@@ -84,79 +89,38 @@ public class RedisCacheSource extends AbstractRedisSource {
}
private void initClient(AnyValue conf) {
- String password = null;
- int urlmaxconns = Utility.cpus();
- int urlpipelines = org.redkale.net.client.Client.DEFAULT_MAX_PIPELINES;
- for (AnyValue node : getNodes(conf)) {
- String urluser = "";
- String urlpwd = "";
- String urldb = "";
- String addrstr = node.getValue(CACHE_SOURCE_URL, node.getValue("addr")); //兼容addr
- if (addrstr.startsWith("redis://")) { //兼容 redis://:1234@127.0.0.1:6379?db=2
- URI uri = URI.create(addrstr);
- address = new InetSocketAddress(uri.getHost(), uri.getPort() > 0 ? uri.getPort() : 6379);
- String userInfo = uri.getUserInfo();
- if (userInfo == null || userInfo.isEmpty()) {
- String authority = uri.getAuthority();
- if (authority != null && authority.indexOf('@') > 0) {
- userInfo = authority.substring(0, authority.indexOf('@'));
- }
- }
- if (userInfo != null && !userInfo.isEmpty()) {
- urlpwd = userInfo;
- if (urlpwd.startsWith(":")) {
- urlpwd = urlpwd.substring(1);
- } else {
- int index = urlpwd.indexOf(':');
- if (index > 0) {
- urluser = urlpwd.substring(0, index);
- urlpwd = urlpwd.substring(index + 1);
- }
- }
- }
- if (uri.getQuery() != null && !uri.getQuery().isEmpty()) {
- String[] qrys = uri.getQuery().split("&|=");
- for (int i = 0; i < qrys.length; i += 2) {
- if (CACHE_SOURCE_USER.equals(qrys[i])) {
- urluser = i == qrys.length - 1 ? "" : qrys[i + 1];
- } else if (CACHE_SOURCE_PASSWORD.equals(qrys[i])) {
- urlpwd = i == qrys.length - 1 ? "" : qrys[i + 1];
- } else if (CACHE_SOURCE_DB.equals(qrys[i])) {
- urldb = i == qrys.length - 1 ? "" : qrys[i + 1];
- } else if (CACHE_SOURCE_MAXCONNS.equals(qrys[i])) {
- urlmaxconns = i == qrys.length - 1 ? Utility.cpus() : Integer.parseInt(qrys[i + 1]);
- } else if (CACHE_SOURCE_PIPELINES.equals(qrys[i])) {
- urlpipelines = i == qrys.length - 1 ? org.redkale.net.client.Client.DEFAULT_MAX_PIPELINES : Integer.parseInt(qrys[i + 1]);
- }
- }
- }
- } else { //兼容addr和port分开
- address = new InetSocketAddress(addrstr, node.getIntValue("port"));
- }
- password = node.getValue(CACHE_SOURCE_PASSWORD, urlpwd).trim();
- String db0 = node.getValue(CACHE_SOURCE_DB, urldb).trim();
- if (!db0.isEmpty()) {
- db = Integer.valueOf(db0);
- }
- break;
+ RedisConfig config = RedisConfig.create(conf);
+ if (config.getAddresses().size() != 1) {
+ throw new RedkaleException("Only one address supported for " + getClass().getSimpleName());
+ }
+ String oneAddr = config.getAddresses().get(0);
+ if (oneAddr.contains("://")) {
+ URI uri = URI.create(oneAddr);
+ address = new InetSocketAddress(uri.getHost(), uri.getPort() > 0 ? uri.getPort() : 6379);
+ } else {
+ int pos = oneAddr.indexOf(':');
+ address = new InetSocketAddress(pos < 0 ? oneAddr : oneAddr.substring(0, pos), pos < 0 ? 6379 : Integer.parseInt(oneAddr.substring(pos + 1)));
}
AsyncGroup ioGroup = clientAsyncGroup;
if (clientAsyncGroup == null) {
String f = "Redkalex-Redis-IOThread-" + resourceName() + "-%s";
ioGroup = AsyncGroup.create(f, workExecutor, 16 * 1024, Utility.cpus() * 4).start();
}
- int maxconns = conf.getIntValue(CACHE_SOURCE_MAXCONNS, urlmaxconns);
- int pipelines = conf.getIntValue(CACHE_SOURCE_PIPELINES, urlpipelines);
RedisCacheClient old = this.client;
-
- if (passwordDecoder != null) {
- password = passwordDecoder.apply(password);
+ this.client = new RedisCacheClient(appName, resourceName(), ioGroup, resourceName() + "." + config.getDb(),
+ new ClientAddress(address), config.getMaxconns(Runtime.getRuntime().availableProcessors()), config.getPipelines(),
+ isEmpty(config.getPassword()) ? null : new RedisCacheReqAuth(config.getPassword()),
+ config.getDb() < 1 ? null : new RedisCacheReqDB(config.getDb()));
+ if (this.subConn != null) {
+ this.subConn.dispose(null);
+ this.subConn = null;
}
- this.client = new RedisCacheClient(resourceName(), ioGroup, resourceName() + "." + db, new ClientAddress(address), maxconns, pipelines,
- password == null || password.isEmpty() ? null : new RedisCacheReqAuth(password), db > 0 ? new RedisCacheReqDB(db) : null);
if (old != null) {
old.close();
}
+ if (!pubsubListeners.isEmpty()) {
+ subConn().join();
+ }
//if (logger.isLoggable(Level.FINE)) logger.log(Level.FINE, RedisCacheSource.class.getSimpleName() + ": addr=" + address + ", db=" + db);
}
@@ -170,46 +134,12 @@ public class RedisCacheSource extends AbstractRedisSource {
for (ResourceEvent event : events) {
sb.append("CacheSource(name=").append(resourceName()).append(") change '").append(event.name()).append("' to '").append(event.coverNewValue()).append("'\r\n");
}
- initClient(this.config);
+ initClient(this.conf);
if (sb.length() > 0) {
logger.log(Level.INFO, sb.toString());
}
}
- public boolean acceptsConf(AnyValue config) {
- if (config == null) {
- return false;
- }
- AnyValue[] nodes = getNodes(config);
- if (nodes == null || nodes.length == 0) {
- return false;
- }
- for (AnyValue node : nodes) {
- String val = node.getValue(CACHE_SOURCE_URL, node.getValue("addr")); //兼容addr
- if (val != null && val.startsWith("redis://")) {
- return true;
- }
- }
- return false;
- }
-
- protected AnyValue[] getNodes(AnyValue config) {
- AnyValue[] nodes = config.getAnyValues(CACHE_SOURCE_NODE);
- if (nodes == null || nodes.length == 0) {
- AnyValue one = config.getAnyValue(CACHE_SOURCE_NODE);
- if (one == null) {
- String val = config.getValue(CACHE_SOURCE_URL);
- if (val == null) {
- return nodes;
- }
- nodes = new AnyValue[]{config};
- } else {
- nodes = new AnyValue[]{one};
- }
- }
- return nodes;
- }
-
@Override
public final String getType() {
return "redis";
@@ -228,578 +158,296 @@ public class RedisCacheSource extends AbstractRedisSource {
}
}
- //--------------------- exists ------------------------------
- @Override
- public CompletableFuture existsAsync(String key) {
- return sendAsync("EXISTS", key, key.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getIntValue(0) > 0);
+ protected CompletableFuture subConn() {
+ RedisCacheConnection conn = this.subConn;
+ if (conn != null) {
+ return CompletableFuture.completedFuture(conn);
+ }
+ return client.newConnection().thenApply(r -> {
+ pubsubLock.lock();
+ try {
+ if (subConn == null) {
+ subConn = r;
+ r.getCodec().withMessageListener(new ClientMessageListener() {
+ @Override
+ public void onMessage(ClientConnection conn, ClientResponse resp) {
+ if (resp.getCause() == null) {
+ RedisCacheResult result = (RedisCacheResult) resp.getMessage();
+ if (result.getFrameValue() == null) {
+ List events = result.getListValue(null, null, byte[].class);
+ String type = new String(events.get(0), StandardCharsets.UTF_8);
+ if (events.size() == 3 && "message".equals(type)) {
+ String channel = new String(events.get(1), StandardCharsets.UTF_8);
+ Set> set = pubsubListeners.get(channel);
+ if (set != null) {
+ byte[] msg = events.get(2);
+ for (CacheEventListener item : set) {
+ subExecutor().execute(() -> {
+ try {
+ item.onMessage(channel, msg);
+ } catch (Throwable t) {
+ logger.log(Level.SEVERE, "CacheSource subscribe message error, topic: " + channel, t);
+ }
+ });
+ }
+ }
+ } else {
+ RedisCacheRequest request = ((RedisCacheCodec) conn.getCodec()).nextRequest();
+ ClientFuture respFuture = ((RedisCacheConnection) conn).pollRespFuture(request.getRequestid());
+ respFuture.complete(result);
+ }
+ } else {
+ RedisCacheRequest request = ((RedisCacheCodec) conn.getCodec()).nextRequest();
+ ClientFuture respFuture = ((RedisCacheConnection) conn).pollRespFuture(request.getRequestid());
+ respFuture.complete(result);
+ }
+ } else {
+ RedisCacheRequest request = ((RedisCacheCodec) conn.getCodec()).nextRequest();
+ ClientFuture respFuture = ((RedisCacheConnection) conn).pollRespFuture(request.getRequestid());
+ respFuture.completeExceptionally(resp.getCause());
+ }
+ }
+
+ public void onClose(ClientConnection conn) {
+ subConn = null;
+ }
+ });
+ //重连时重新订阅
+ if (!pubsubListeners.isEmpty()) {
+ final Map, HashSet> listeners = new HashMap<>();
+ pubsubListeners.forEach((t, s) -> {
+ s.forEach(l -> listeners.computeIfAbsent(l, x -> new HashSet<>()).add(t));
+ });
+ listeners.forEach((listener, topics) -> {
+ subscribeAsync(listener, topics.toArray(Creator.funcStringArray()));
+ });
+ }
+ }
+ return subConn;
+ } finally {
+ pubsubLock.unlock();
+ }
+ });
}
@Override
- public boolean exists(String key) {
- return existsAsync(key).join();
+ public CompletableFuture isOpenAsync() {
+ return CompletableFuture.completedFuture(client != null);
+ }
+
+ //------------------------ 订阅发布 SUB/PUB ------------------------
+ @Override
+ public CompletableFuture> pubsubChannelsAsync(@Nullable String pattern) {
+ CompletableFuture future = pattern == null ? sendAsync(RedisCommand.PUBSUB, "CHANNELS", CHANNELS)
+ : sendAsync(RedisCommand.PUBSUB, "CHANNELS", CHANNELS, pattern.getBytes(StandardCharsets.UTF_8));
+ return future.thenApply(v -> v.getListValue("CHANNELS", null, String.class));
+ }
+
+ @Override
+ public CompletableFuture subscribeAsync(CacheEventListener listener, String... topics) {
+ Objects.requireNonNull(listener);
+ if (topics == null || topics.length < 1) {
+ throw new RedkaleException("topics is empty");
+ }
+ WorkThread workThread = WorkThread.currentWorkThread();
+ String traceid = Traces.currentTraceid();
+ return subConn()
+ .thenCompose(conn
+ -> conn.writeRequest(conn.pollRequest(workThread, traceid).prepare(RedisCommand.SUBSCRIBE, null, keysArgs(topics)))
+ .thenApply(v -> {
+ for (String topic : topics) {
+ pubsubListeners.computeIfAbsent(topic, y -> new CopyOnWriteArraySet<>()).add(listener);
+ }
+ return null;
+ })
+ );
+ }
+
+ @Override
+ public CompletableFuture unsubscribeAsync(CacheEventListener listener, String... topics) {
+ if (listener == null) { //清掉指定topic的所有订阅者
+ Set delTopics = new HashSet<>();
+ if (topics == null || topics.length < 1) {
+ delTopics.addAll(pubsubListeners.keySet());
+ } else {
+ delTopics.addAll(Arrays.asList(topics));
+ }
+ List> futures = new ArrayList<>();
+ delTopics.forEach(topic -> {
+ futures.add(subConn().thenCompose(conn -> conn.writeRequest(RedisCacheRequest.create(RedisCommand.UNSUBSCRIBE, topic, topic.getBytes(StandardCharsets.UTF_8)))
+ .thenApply(r -> {
+ pubsubListeners.remove(topic);
+ return null;
+ })
+ ));
+ });
+ return returnFutureSize(futures);
+ } else { //清掉指定topic的指定订阅者
+ List> futures = new ArrayList<>();
+ for (String topic : topics) {
+ CopyOnWriteArraySet> listens = pubsubListeners.get(topic);
+ if (listens == null) {
+ continue;
+ }
+ listens.remove(listener);
+ if (listens.isEmpty()) {
+ futures.add(subConn().thenCompose(conn -> conn.writeRequest(RedisCacheRequest.create(RedisCommand.UNSUBSCRIBE, topic, topic.getBytes(StandardCharsets.UTF_8)))
+ .thenApply(r -> {
+ pubsubListeners.remove(topic);
+ return null;
+ })
+ ));
+ }
+ }
+ return returnFutureSize(futures);
+ }
+ }
+
+ @Override
+ public CompletableFuture publishAsync(String topic, byte[] message) {
+ Objects.requireNonNull(topic);
+ Objects.requireNonNull(message);
+ return sendAsync(RedisCommand.PUBLISH, topic, topic.getBytes(StandardCharsets.UTF_8), message).thenApply(v -> v.getIntValue(0));
+ }
+
+ //--------------------- exists ------------------------------
+ @Override
+ public CompletableFuture existsAsync(String key) {
+ return sendAsync(RedisCommand.EXISTS, key, keyArgs(key)).thenApply(v -> v.getIntValue(0) > 0);
}
//--------------------- get ------------------------------
@Override
public CompletableFuture getAsync(String key, Type type) {
- return sendAsync("GET", key, key.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getObjectValue(key, cryptor, type));
- }
-
- @Override
- public CompletableFuture getStringAsync(String key) {
- return sendAsync("GET", key, key.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getStringValue(key, cryptor));
- }
-
- @Override
- public CompletableFuture getSetStringAsync(String key, String value) {
- return sendAsync("GETSET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getStringValue(key, cryptor));
- }
-
- @Override
- public CompletableFuture getLongAsync(String key, long defValue) {
- return sendAsync("GET", key, key.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getLongValue(defValue));
- }
-
- @Override
- public CompletableFuture getSetLongAsync(String key, long value, long defValue) {
- return sendAsync("GETSET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(value)).thenApply(v -> v.getLongValue(defValue));
- }
-
- @Override
- public T get(String key, final Type type) {
- return (T) getAsync(key, type).join();
- }
-
- @Override
- public String getString(String key) {
- return getStringAsync(key).join();
- }
-
- @Override
- public String getSetString(String key, String value) {
- return getSetStringAsync(key, value).join();
- }
-
- @Override
- public long getLong(String key, long defValue) {
- return getLongAsync(key, defValue).join();
- }
-
- @Override
- public long getSetLong(String key, long value, long defValue) {
- return getSetLongAsync(key, value, defValue).join();
+ return sendAsync(RedisCommand.GET, key, keyArgs(key)).thenApply(v -> v.getObjectValue(key, cryptor, type));
}
//--------------------- getex ------------------------------
@Override
public CompletableFuture getexAsync(String key, int expireSeconds, final Type type) {
- return sendAsync("GETEX", key, key.getBytes(StandardCharsets.UTF_8), "EX".getBytes(StandardCharsets.UTF_8), String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getObjectValue(key, cryptor, type));
+ return sendAsync(RedisCommand.GETEX, key, keyArgs(key, "EX", expireSeconds)).thenApply(v -> v.getObjectValue(key, cryptor, type));
}
@Override
- public T getex(String key, final int expireSeconds, final Type type) {
- return (T) getexAsync(key, expireSeconds, type).join();
- }
-
- @Override
- public CompletableFuture getexStringAsync(String key, int expireSeconds) {
- return sendAsync("GETEX", key, key.getBytes(StandardCharsets.UTF_8), "EX".getBytes(StandardCharsets.UTF_8), String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getStringValue(key, cryptor));
- }
-
- @Override
- public String getexString(String key, final int expireSeconds) {
- return getexStringAsync(key, expireSeconds).join();
- }
-
- @Override
- public CompletableFuture getexLongAsync(String key, int expireSeconds, long defValue) {
- return sendAsync("GETEX", key, key.getBytes(StandardCharsets.UTF_8), "EX".getBytes(StandardCharsets.UTF_8), String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getLongValue(defValue));
- }
-
- @Override
- public long getexLong(String key, final int expireSeconds, long defValue) {
- return getexLongAsync(key, expireSeconds, defValue).join();
- }
-
- @Override
- public CompletableFuture getexBytesAsync(final String key, final int expireSeconds) {
- return sendAsync("GETEX", key, key.getBytes(StandardCharsets.UTF_8), "EX".getBytes(StandardCharsets.UTF_8), String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getFrameValue());
- }
-
- @Override
- public byte[] getexBytes(final String key, final int expireSeconds) {
- return getexBytesAsync(key, expireSeconds).join();
- }
-
- @Override
- public CompletableFuture msetAsync(final Object... keyVals) {
+ public CompletableFuture msetAsync(final Serializable... keyVals) {
if (keyVals.length % 2 != 0) {
throw new RedkaleException("key value must be paired");
}
- byte[][] bs = new byte[keyVals.length][];
- for (int i = 0; i < keyVals.length; i += 2) {
- String key = keyVals[i].toString();
- bs[i] = key.getBytes(StandardCharsets.UTF_8);
- bs[i + 1] = formatValue(key, cryptor, keyVals[i + 1]);
- }
- return sendAsync("MSET", keyVals[0].toString(), bs).thenApply(v -> v.getVoidValue());
+ return sendAsync(RedisCommand.MSET, keyVals[0].toString(), keymArgs(keyVals)).thenApply(v -> v.getVoidValue());
}
@Override
public CompletableFuture msetAsync(final Map map) {
- if (map == null || map.isEmpty()) {
+ if (isEmpty(map)) {
return CompletableFuture.completedFuture(null);
}
- List bs = new ArrayList<>();
- StringWrapper onekey = new StringWrapper();
- map.forEach((key, val) -> {
- onekey.setValue(key.toString());
- bs.add(key.toString().getBytes(StandardCharsets.UTF_8));
- bs.add(formatValue(key.toString(), cryptor, val));
- });
- return sendAsync("MSET", onekey.getValue(), bs.toArray(new byte[bs.size()][])).thenApply(v -> v.getVoidValue());
+ return sendAsync(RedisCommand.MSET, map.keySet().stream().findFirst().orElse("").toString(), keymArgs(map)).thenApply(v -> v.getVoidValue());
}
//--------------------- setex ------------------------------
- @Override
- public CompletableFuture setAsync(String key, final Type type, T value) {
- return sendAsync("SET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, (Convert) null, type, value)).thenApply(v -> v.getVoidValue());
- }
-
@Override
public CompletableFuture setAsync(String key, Convert convert, final Type type, T value) {
- return sendAsync("SET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, convert, type, value)).thenApply(v -> v.getVoidValue());
- }
-
- @Override
- public CompletableFuture setnxAsync(String key, final Type type, T value) {
- return sendAsync("SETNX", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, (Convert) null, type, value)).thenApply(v -> v.getBoolValue());
+ return sendAsync(RedisCommand.SET, key, keyArgs(key, convert, type, value)).thenApply(v -> v.getVoidValue());
}
@Override
public CompletableFuture setnxAsync(String key, Convert convert, final Type type, T value) {
- return sendAsync("SETNX", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, convert, type, value)).thenApply(v -> v.getBoolValue());
- }
-
- @Override
- public boolean setnxBytes(final String key, final byte[] value) {
- return setnxBytesAsync(key, value).join();
- }
-
- @Override
- public CompletableFuture setnxBytesAsync(final String key, byte[] value) {
- return sendAsync("SETNX", key, key.getBytes(StandardCharsets.UTF_8), value).thenApply(v -> v.getBoolValue());
- }
-
- @Override
- public CompletableFuture getSetAsync(String key, final Type type, T value) {
- return sendAsync("GETSET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, (Convert) null, type, value)).thenApply(v -> v.getObjectValue(key, cryptor, type));
+ return sendAsync(RedisCommand.SETNX, key, keyArgs(key, convert, type, value)).thenApply(v -> v.getBoolValue());
}
@Override
public CompletableFuture getSetAsync(String key, Convert convert, final Type type, T value) {
- return sendAsync("GETSET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, convert, type, value)).thenApply(v -> v.getObjectValue(key, cryptor, type));
+ return sendAsync(RedisCommand.GETSET, key, keyArgs(key, convert, type, value)).thenApply(v -> v.getObjectValue(key, cryptor, type));
}
@Override
- public void mset(final Object... keyVals) {
- msetAsync(keyVals).join();
- }
-
- @Override
- public void mset(final Map map) {
- msetAsync(map).join();
- }
-
- @Override
- public void set(final String key, final Type type, T value) {
- setAsync(key, type, value).join();
- }
-
- @Override
- public void set(String key, final Convert convert, final Type type, T value) {
- setAsync(key, convert, type, value).join();
- }
-
- @Override
- public boolean setnx(final String key, final Type type, T value) {
- return setnxAsync(key, type, value).join();
- }
-
- @Override
- public boolean setnx(String key, final Convert convert, final Type type, T value) {
- return setnxAsync(key, convert, type, value).join();
- }
-
- @Override
- public T getSet(String key, final Type type, T value) {
- return getSetAsync(key, type, value).join();
- }
-
- @Override
- public T getSet(String key, Convert convert, final Type type, T value) {
- return getSetAsync(key, convert, type, value).join();
- }
-
- @Override
- public CompletableFuture setStringAsync(String key, String value) {
- return sendAsync("SET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getVoidValue());
- }
-
- @Override
- public CompletableFuture setnxStringAsync(String key, String value) {
- return sendAsync("SETNX", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getBoolValue());
- }
-
- @Override
- public void setString(String key, String value) {
- setStringAsync(key, value).join();
- }
-
- @Override
- public boolean setnxString(String key, String value) {
- return setnxStringAsync(key, value).join();
- }
-
- @Override
- public CompletableFuture setLongAsync(String key, long value) {
- return sendAsync("SET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getVoidValue());
- }
-
- @Override
- public CompletableFuture setnxLongAsync(String key, long value) {
- return sendAsync("SETNX", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getBoolValue());
- }
-
- @Override
- public void setLong(String key, long value) {
- setLongAsync(key, value).join();
- }
-
- @Override
- public boolean setnxLong(String key, long value) {
- return setnxLongAsync(key, value).join();
+ public CompletableFuture getDelAsync(String key, final Type type) {
+ return sendAsync(RedisCommand.GETDEL, key, keyArgs(key)).thenApply(v -> v.getObjectValue(key, cryptor, type));
}
//--------------------- setex ------------------------------
- @Override
- public CompletableFuture setexAsync(String key, int expireSeconds, final Type type, T value) {
- return sendAsync("SETEX", key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, (Convert) null, type, value)).thenApply(v -> v.getVoidValue());
- }
-
@Override
public CompletableFuture setexAsync(String key, int expireSeconds, Convert convert, final Type type, T value) {
- return sendAsync("SETEX", key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, convert, type, value)).thenApply(v -> v.getVoidValue());
- }
-
- @Override
- public void setex(String key, int expireSeconds, final Type type, T value) {
- setexAsync(key, expireSeconds, type, value).join();
- }
-
- @Override
- public void setex(String key, int expireSeconds, Convert convert, final Type type, T value) {
- setexAsync(key, expireSeconds, convert, type, value).join();
- }
-
- @Override
- public CompletableFuture setexStringAsync(String key, int expireSeconds, String value) {
- return sendAsync("SETEX", key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getVoidValue());
- }
-
- @Override
- public void setexString(String key, int expireSeconds, String value) {
- setexStringAsync(key, expireSeconds, value).join();
- }
-
- @Override
- public CompletableFuture setexLongAsync(String key, int expireSeconds, long value) {
- return sendAsync("SETEX", key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getVoidValue());
- }
-
- @Override
- public void setexLong(String key, int expireSeconds, long value) {
- setexLongAsync(key, expireSeconds, value).join();
- }
-
- @Override
- public CompletableFuture setnxexStringAsync(String key, int expireSeconds, String value) {
- return sendAsync("SET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value), NX, EX, String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getBoolValue());
- }
-
- @Override
- public CompletableFuture setnxexLongAsync(String key, int expireSeconds, long value) {
- return sendAsync("SET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value), NX, EX, String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getBoolValue());
- }
-
- @Override
- public CompletableFuture setnxexBytesAsync(String key, int expireSeconds, byte[] value) {
- return sendAsync("SET", key, key.getBytes(StandardCharsets.UTF_8), value, NX, EX, String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getBoolValue());
- }
-
- @Override
- public CompletableFuture setnxexAsync(String key, int expireSeconds, final Type type, T value) {
- return sendAsync("SET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, (Convert) null, type, value), NX, EX, String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getBoolValue());
+ return sendAsync(RedisCommand.SETEX, key, keyArgs(key, expireSeconds, convert, type, value)).thenApply(v -> v.getVoidValue());
}
@Override
public CompletableFuture setnxexAsync(String key, int expireSeconds, Convert convert, final Type type, T value) {
- return sendAsync("SET", key, key.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, convert, type, value), NX, EX, String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getBoolValue());
- }
-
- @Override
- public boolean setnxex(final String key, final int expireSeconds, final Type type, final T value) {
- return setnxexAsync(key, expireSeconds, type, value).join();
- }
-
- @Override
- public boolean setnxex(final String key, final int expireSeconds, final Convert convert, final Type type, final T value) {
- return setnxexAsync(key, expireSeconds, convert, type, value).join();
- }
-
- @Override
- public boolean setnxexString(final String key, final int expireSeconds, final String value) {
- return setnxexStringAsync(key, expireSeconds, value).join();
- }
-
- @Override
- public boolean setnxexLong(final String key, final int expireSeconds, final long value) {
- return setnxexLongAsync(key, expireSeconds, value).join();
- }
-
- @Override
- public boolean setnxexBytes(final String key, final int expireSeconds, final byte[] value) {
- return setnxexBytesAsync(key, expireSeconds, value).join();
+ return sendAsync(RedisCommand.SET, key, keyArgs(key, expireSeconds, NX, EX, convert, type, value)).thenApply(v -> v.getBoolValue());
}
//--------------------- expire ------------------------------
@Override
public CompletableFuture expireAsync(String key, int expireSeconds) {
- return sendAsync("EXPIRE", key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getVoidValue());
+ return sendAsync(RedisCommand.EXPIRE, key, keyArgs(key, expireSeconds)).thenApply(v -> v.getVoidValue());
+ }
+
+ //--------------------- persist ------------------------------
+ @Override
+ public CompletableFuture persistAsync(String key) {
+ return sendAsync(RedisCommand.PERSIST, key, keyArgs(key)).thenApply(v -> v.getBoolValue());
+ }
+
+ //--------------------- rename ------------------------------
+ @Override
+ public CompletableFuture renameAsync(String oldKey, String newKey) {
+ return sendAsync(RedisCommand.RENAME, oldKey, keysArgs(oldKey, newKey)).thenApply(v -> v.getBoolValue());
}
@Override
- public void expire(String key, int expireSeconds) {
- expireAsync(key, expireSeconds).join();
+ public CompletableFuture renamenxAsync(String oldKey, String newKey) {
+ return sendAsync(RedisCommand.RENAMENX, oldKey, keysArgs(oldKey, newKey)).thenApply(v -> v.getBoolValue());
}
//--------------------- del ------------------------------
@Override
- public CompletableFuture delAsync(String... keys) {
+ public CompletableFuture delAsync(String... keys) {
if (keys.length == 0) {
- return CompletableFuture.completedFuture(0);
+ return CompletableFuture.completedFuture(0L);
}
- if (keys.length == 1) {
- return sendAsync("DEL", keys[0], keys[0].getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getIntValue(0));
- } else {
- byte[][] bs = new byte[keys.length][];
- for (int i = 0; i < keys.length; i++) {
- bs[i] = keys[i].getBytes(StandardCharsets.UTF_8);
- }
- return sendAsync("DEL", keys[0], bs).thenApply(v -> v.getIntValue(0));
- }
- }
-
- @Override
- public int del(String... keys) {
- return delAsync(keys).join();
+ return sendAsync(RedisCommand.DEL, keys[0], keysArgs(keys)).thenApply(v -> v.getLongValue(0L));
}
//--------------------- incrby ------------------------------
- @Override
- public long incr(final String key) {
- return incrAsync(key).join();
- }
-
@Override
public CompletableFuture incrAsync(final String key) {
- return sendAsync("INCR", key, key.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getLongValue(0L));
- }
-
- @Override
- public long incrby(final String key, long num) {
- return incrbyAsync(key, num).join();
- }
-
- @Override
- public double incrbyFloat(final String key, double num) {
- return incrbyFloatAsync(key, num).join();
+ return sendAsync(RedisCommand.INCR, key, keyArgs(key)).thenApply(v -> v.getLongValue(0L));
}
@Override
public CompletableFuture incrbyAsync(final String key, long num) {
- return sendAsync("INCRBY", key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(num).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getLongValue(0L));
+ return sendAsync(RedisCommand.INCRBY, key, keyArgs(key, num)).thenApply(v -> v.getLongValue(0L));
}
@Override
public CompletableFuture incrbyFloatAsync(final String key, double num) {
- return sendAsync("INCRBYFLOAT", key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(num).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getDoubleValue(0.d));
+ return sendAsync(RedisCommand.INCRBYFLOAT, key, keyArgs(key, num)).thenApply(v -> v.getDoubleValue(0.d));
}
//--------------------- decrby ------------------------------
- @Override
- public long decr(final String key) {
- return decrAsync(key).join();
- }
-
@Override
public CompletableFuture decrAsync(final String key) {
- return sendAsync("DECR", key, key.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getLongValue(0L));
- }
-
- @Override
- public long decrby(final String key, long num) {
- return decrbyAsync(key, num).join();
+ return sendAsync(RedisCommand.DECR, key, keyArgs(key)).thenApply(v -> v.getLongValue(0L));
}
@Override
public CompletableFuture decrbyAsync(final String key, long num) {
- return sendAsync("DECRBY", key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(num).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getLongValue(0L));
+ return sendAsync(RedisCommand.DECRBY, key, keyArgs(key, num)).thenApply(v -> v.getLongValue(0L));
}
@Override
- public int hdel(final String key, String... fields) {
- return hdelAsync(key, fields).join();
+ public CompletableFuture hdelAsync(final String key, String... fields) {
+ return sendAsync(RedisCommand.HDEL, key, keysArgs(key, fields)).thenApply(v -> v.getLongValue(0L));
}
@Override
- public int hlen(final String key) {
- return hlenAsync(key).join();
- }
-
- @Override
- public List hkeys(final String key) {
- return hkeysAsync(key).join();
- }
-
- @Override
- public long hincr(final String key, String field) {
- return hincrAsync(key, field).join();
- }
-
- @Override
- public long hincrby(final String key, String field, long num) {
- return hincrbyAsync(key, field, num).join();
- }
-
- @Override
- public double hincrbyFloat(final String key, String field, double num) {
- return hincrbyFloatAsync(key, field, num).join();
- }
-
- @Override
- public long hdecr(final String key, String field) {
- return hdecrAsync(key, field).join();
- }
-
- @Override
- public long hdecrby(final String key, String field, long num) {
- return hdecrbyAsync(key, field, num).join();
- }
-
- @Override
- public boolean hexists(final String key, String field) {
- return hexistsAsync(key, field).join();
- }
-
- @Override
- public void hset(final String key, final String field, final Type type, final T value) {
- hsetAsync(key, field, type, value).join();
- }
-
- @Override
- public void hset(final String key, final String field, final Convert convert, final Type type, final T value) {
- hsetAsync(key, field, convert, type, value).join();
- }
-
- @Override
- public void hsetString(final String key, final String field, final String value) {
- hsetStringAsync(key, field, value).join();
- }
-
- @Override
- public void hsetLong(final String key, final String field, final long value) {
- hsetLongAsync(key, field, value).join();
- }
-
- @Override
- public boolean hsetnx(final String key, final String field, final Type type, final T value) {
- return hsetnxAsync(key, field, type, value).join();
- }
-
- @Override
- public boolean hsetnx(final String key, final String field, final Convert convert, final Type type, final T value) {
- return hsetnxAsync(key, field, convert, type, value).join();
- }
-
- @Override
- public boolean hsetnxString(final String key, final String field, final String value) {
- return hsetnxStringAsync(key, field, value).join();
- }
-
- @Override
- public boolean hsetnxLong(final String key, final String field, final long value) {
- return hsetnxLongAsync(key, field, value).join();
- }
-
- @Override
- public void hmset(final String key, final Serializable... values) {
- hmsetAsync(key, values).join();
- }
-
- @Override
- public void hmset(final String key, final Map map) {
- hmsetAsync(key, map).join();
- }
-
- @Override
- public List hmget(final String key, final Type type, final String... fields) {
- return hmgetAsync(key, type, fields).join();
- }
-
- @Override
- public Map hmap(final String key, final Type type, int offset, int limit, String pattern) {
- return (Map) hmapAsync(key, type, offset, limit, pattern).join();
- }
-
- @Override
- public Map hmap(final String key, final Type type, int offset, int limit) {
- return (Map) hmapAsync(key, type, offset, limit).join();
- }
-
- @Override
- public T hget(final String key, final String field, final Type type) {
- return (T) hgetAsync(key, field, type).join();
- }
-
- @Override
- public String hgetString(final String key, final String field) {
- return hgetStringAsync(key, field).join();
- }
-
- @Override
- public long hgetLong(final String key, final String field, long defValue) {
- return hgetLongAsync(key, field, defValue).join();
- }
-
- @Override
- public CompletableFuture hdelAsync(final String key, String... fields) {
- byte[][] bs = new byte[fields.length + 1][];
- bs[0] = key.getBytes(StandardCharsets.UTF_8);
- for (int i = 0; i < fields.length; i++) {
- bs[i + 1] = fields[i].getBytes(StandardCharsets.UTF_8);
- }
- return sendAsync("HDEL", key, bs).thenApply(v -> v.getIntValue(0));
- }
-
- @Override
- public CompletableFuture hlenAsync(final String key) {
- return sendAsync("HLEN", key, key.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getIntValue(0));
+ public CompletableFuture hlenAsync(final String key) {
+ return sendAsync(RedisCommand.HLEN, key, keyArgs(key)).thenApply(v -> v.getLongValue(0L));
}
@Override
public CompletableFuture> hkeysAsync(final String key) {
- return sendAsync("HKEYS", key, key.getBytes(StandardCharsets.UTF_8)).thenApply(v -> (List) v.getListValue(key, cryptor, String.class));
+ return sendAsync(RedisCommand.HKEYS, key, keyArgs(key)).thenApply(v -> (List) v.getListValue(key, cryptor, String.class));
}
@Override
@@ -809,12 +457,12 @@ public class RedisCacheSource extends AbstractRedisSource {
@Override
public CompletableFuture hincrbyAsync(final String key, String field, long num) {
- return sendAsync("HINCRBY", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), String.valueOf(num).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getLongValue(0L));
+ return sendAsync(RedisCommand.HINCRBY, key, keysArgs(key, field, String.valueOf(num))).thenApply(v -> v.getLongValue(0L));
}
@Override
public CompletableFuture hincrbyFloatAsync(final String key, String field, double num) {
- return sendAsync("HINCRBYFLOAT", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), String.valueOf(num).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getDoubleValue(0.d));
+ return sendAsync(RedisCommand.HINCRBYFLOAT, key, keysArgs(key, field, String.valueOf(num))).thenApply(v -> v.getDoubleValue(0.d));
}
@Override
@@ -829,15 +477,7 @@ public class RedisCacheSource extends AbstractRedisSource {
@Override
public CompletableFuture hexistsAsync(final String key, String field) {
- return sendAsync("HEXISTS", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getIntValue(0) > 0);
- }
-
- @Override
- public CompletableFuture hsetAsync(final String key, final String field, final Type type, final T value) {
- if (value == null) {
- return CompletableFuture.completedFuture(null);
- }
- return sendAsync("HSET", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, null, type, value)).thenApply(v -> v.getVoidValue());
+ return sendAsync(RedisCommand.HEXISTS, key, keysArgs(key, field)).thenApply(v -> v.getIntValue(0) > 0);
}
@Override
@@ -845,28 +485,7 @@ public class RedisCacheSource extends AbstractRedisSource {
if (value == null) {
return CompletableFuture.completedFuture(null);
}
- return sendAsync("HSET", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, convert, type, value)).thenApply(v -> v.getVoidValue());
- }
-
- @Override
- public CompletableFuture hsetStringAsync(final String key, final String field, final String value) {
- if (value == null) {
- return CompletableFuture.completedFuture(null);
- }
- return sendAsync("HSET", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getVoidValue());
- }
-
- @Override
- public CompletableFuture hsetLongAsync(final String key, final String field, final long value) {
- return sendAsync("HSET", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getVoidValue());
- }
-
- @Override
- public CompletableFuture hsetnxAsync(final String key, final String field, final Type type, final T value) {
- if (value == null) {
- return CompletableFuture.completedFuture(null);
- }
- return sendAsync("HSETNX", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, null, type, value)).thenApply(v -> v.getBoolValue());
+ return sendAsync(RedisCommand.HSET, key, keyArgs(key, field, convert, type, value)).thenApply(v -> v.getVoidValue());
}
@Override
@@ -874,628 +493,594 @@ public class RedisCacheSource extends AbstractRedisSource {
if (value == null) {
return CompletableFuture.completedFuture(null);
}
- return sendAsync("HSETNX", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, convert, type, value)).thenApply(v -> v.getBoolValue());
- }
-
- @Override
- public CompletableFuture hsetnxStringAsync(final String key, final String field, final String value) {
- if (value == null) {
- return CompletableFuture.completedFuture(false);
- }
- return sendAsync("HSETNX", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getBoolValue());
- }
-
- @Override
- public CompletableFuture hsetnxLongAsync(final String key, final String field, final long value) {
- return sendAsync("HSETNX", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(key, cryptor, value)).thenApply(v -> v.getBoolValue());
+ return sendAsync(RedisCommand.HSETNX, key, keyArgs(key, field, convert, type, value)).thenApply(v -> v.getBoolValue());
}
@Override
public CompletableFuture hmsetAsync(final String key, final Serializable... values) {
- byte[][] bs = new byte[values.length + 1][];
- bs[0] = key.getBytes(StandardCharsets.UTF_8);
- for (int i = 0; i < values.length; i += 2) {
- bs[i + 1] = String.valueOf(values[i]).getBytes(StandardCharsets.UTF_8);
- bs[i + 2] = formatValue(key, cryptor, values[i + 1]);
- }
- return sendAsync("HMSET", key, bs).thenApply(v -> v.getVoidValue());
+ return sendAsync(RedisCommand.HMSET, key, keyMapArgs(key, values)).thenApply(v -> v.getVoidValue());
}
@Override
public CompletableFuture hmsetAsync(final String key, final Map map) {
- if (map == null || map.isEmpty()) {
+ if (isEmpty(map)) {
return CompletableFuture.completedFuture(null);
}
- List bs = new ArrayList<>();
- bs.add(key.getBytes(StandardCharsets.UTF_8));
- map.forEach((k, v) -> {
- bs.add(k.toString().getBytes(StandardCharsets.UTF_8));
- bs.add(formatValue(k.toString(), cryptor, v));
- });
- return sendAsync("HMSET", key, bs.toArray(new byte[bs.size()][])).thenApply(v -> v.getVoidValue());
+ return sendAsync(RedisCommand.HMSET, key, keyMapArgs(key, map)).thenApply(v -> v.getVoidValue());
}
@Override
public CompletableFuture> hmgetAsync(final String key, final Type type, final String... fields) {
- byte[][] bs = new byte[fields.length + 1][];
- bs[0] = key.getBytes(StandardCharsets.UTF_8);
- for (int i = 0; i < fields.length; i++) {
- bs[i + 1] = fields[i].getBytes(StandardCharsets.UTF_8);
- }
- return sendAsync("HMGET", key, bs).thenApply(v -> (List) v.getListValue(key, cryptor, type));
+ return sendAsync(RedisCommand.HMGET, key, keysArgs(key, fields)).thenApply(v -> (List) v.getListValue(key, cryptor, type));
}
@Override
- public CompletableFuture> hmapAsync(final String key, final Type type, int offset, int limit) {
- return hmapAsync(key, type, offset, limit, null);
+ public CompletableFuture> hscanAsync(final String key, final Type type, AtomicLong cursor, int limit, String pattern) {
+ return sendAsync(RedisCommand.HSCAN, key, keyArgs(key, cursor, limit, pattern)).thenApply(v -> {
+ Map map = v.getMapValue(key, cryptor, type);
+ cursor.set(v.getCursor());
+ return map;
+ });
}
@Override
- public CompletableFuture> hmapAsync(final String key, final Type type, int offset, int limit, String pattern) {
- byte[][] bs = new byte[pattern == null || pattern.isEmpty() ? 4 : 6][limit];
- int index = -1;
- bs[++index] = key.getBytes(StandardCharsets.UTF_8);
- bs[++index] = String.valueOf(offset).getBytes(StandardCharsets.UTF_8);
- if (pattern != null && !pattern.isEmpty()) {
- bs[++index] = "MATCH".getBytes(StandardCharsets.UTF_8);
- bs[++index] = pattern.getBytes(StandardCharsets.UTF_8);
- }
- bs[++index] = "COUNT".getBytes(StandardCharsets.UTF_8);
- bs[++index] = String.valueOf(limit).getBytes(StandardCharsets.UTF_8);
- return sendAsync("HSCAN", key, bs).thenApply(v -> v.getMapValue(key, cryptor, type));
+ public CompletableFuture> hgetallAsync(final String key, final Type type) {
+ return sendAsync(RedisCommand.HGETALL, key, keyArgs(key)).thenApply(v -> v.getMapValue(key, cryptor, type));
+ }
+
+ @Override
+ public CompletableFuture> hvalsAsync(final String key, final Type type) {
+ return sendAsync(RedisCommand.HVALS, key, keyArgs(key)).thenApply(v -> v.getListValue(key, cryptor, type));
}
@Override
public CompletableFuture hgetAsync(final String key, final String field, final Type type) {
- return sendAsync("HGET", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getObjectValue(key, cryptor, type));
+ return sendAsync(RedisCommand.HGET, key, keysArgs(key, field)).thenApply(v -> v.getObjectValue(key, cryptor, type));
}
@Override
- public CompletableFuture hgetStringAsync(final String key, final String field) {
- return sendAsync("HGET", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getStringValue(key, cryptor));
- }
-
- @Override
- public CompletableFuture hgetLongAsync(final String key, final String field, long defValue) {
- return sendAsync("HGET", key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getLongValue(defValue));
+ public CompletableFuture hstrlenAsync(String key, final String field) {
+ return sendAsync(RedisCommand.HSTRLEN, key, keysArgs(key, field)).thenApply(v -> v.getLongValue(0L));
}
@Override
public CompletableFuture> smembersAsync(String key, final Type componentType) {
- return sendAsync("SMEMBERS", key, keySetArgs(key)).thenApply(v -> v.getSetValue(key, cryptor, componentType));
+ return sendAsync(RedisCommand.SMEMBERS, key, keyArgs(key)).thenApply(v -> v.getSetValue(key, cryptor, componentType));
}
@Override
- public CompletableFuture> lrangeAsync(String key, final Type componentType) {
- return sendAsync("LRANGE", key, keyListArgs(key)).thenApply(v -> v.getListValue(key, cryptor, componentType));
+ public CompletableFuture> lrangeAsync(String key, final Type componentType, int start, int stop) {
+ return sendAsync(RedisCommand.LRANGE, key, keyArgs(key, start, stop)).thenApply(v -> v.getListValue(key, cryptor, componentType));
+ }
+
+ @Override
+ public CompletableFuture lindexAsync(String key, Type componentType, int index) {
+ return sendAsync(RedisCommand.LINDEX, key, keyArgs(key, index)).thenApply(v -> v.getObjectValue(key, cryptor, componentType));
+ }
+
+ @Override
+ public CompletableFuture linsertBeforeAsync(String key, Type componentType, T pivot, T value) {
+ return sendAsync(RedisCommand.LINSERT, key, keyArgs(key, "BEFORE", componentType, pivot, value)).thenApply(v -> v.getLongValue(0L));
+ }
+
+ @Override
+ public CompletableFuture linsertAfterAsync(String key, Type componentType, T pivot, T value) {
+ return sendAsync(RedisCommand.LINSERT, key, keyArgs(key, "AFTER", componentType, pivot, value)).thenApply(v -> v.getLongValue(0L));
+ }
+
+ @Override
+ public CompletableFuture ltrimAsync(final String key, int start, int stop) {
+ return sendAsync(RedisCommand.LTRIM, key, keyArgs(key, start, stop)).thenApply(v -> null);
+ }
+
+ @Override
+ public CompletableFuture lpopAsync(final String key, final Type componentType) {
+ return sendAsync(RedisCommand.LPOP, key, keyArgs(key)).thenApply(v -> v.getObjectValue(key, cryptor, componentType));
+ }
+
+ @Override
+ public CompletableFuture lpushAsync(final String key, final Type componentType, T... values) {
+ return sendAsync(RedisCommand.LPUSH, key, keyArgs(key, componentType, values)).thenApply(v -> null);
+ }
+
+ @Override
+ public CompletableFuture lpushxAsync(final String key, final Type componentType, T... values) {
+ return sendAsync(RedisCommand.LPUSHX, key, keyArgs(key, componentType, values)).thenApply(v -> null);
+ }
+
+ @Override
+ public CompletableFuture rpopAsync(final String key, final Type componentType) {
+ return sendAsync(RedisCommand.RPOP, key, keyArgs(key)).thenApply(v -> v.getObjectValue(key, cryptor, componentType));
+ }
+
+ @Override
+ public CompletableFuture rpoplpushAsync(final String key, final String key2, final Type componentType) {
+ return sendAsync(RedisCommand.RPOPLPUSH, key, keysArgs(key, key2)).thenApply(v -> v.getObjectValue(key, cryptor, componentType));
}
//--------------------- collection ------------------------------
@Override
- public CompletableFuture llenAsync(String key) {
- return sendAsync("LLEN", key, key.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getIntValue(0));
+ public CompletableFuture llenAsync(String key) {
+ return sendAsync(RedisCommand.LLEN, key, keyArgs(key)).thenApply(v -> v.getLongValue(0L));
}
@Override
- public CompletableFuture scardAsync(String key) {
- return sendAsync("SCARD", key, key.getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getIntValue(0));
+ public CompletableFuture scardAsync(String key) {
+ return sendAsync(RedisCommand.SCARD, key, keyArgs(key)).thenApply(v -> v.getLongValue(0L));
}
@Override
- public int llen(String key) {
- return llenAsync(key).join();
+ public CompletableFuture> smismembersAsync(final String key, final String... members) {
+ return sendAsync(RedisCommand.SMISMEMBER, key, keysArgs(key, members)).thenApply(v -> v.getListValue(key, cryptor, Boolean.class));
}
@Override
- public int scard(String key) {
- return scardAsync(key).join();
+ public CompletableFuture smoveAsync(String key, String key2, Type componentType, T member) {
+ return sendAsync(RedisCommand.SMOVE, key, keyArgs(key, key2, componentType, member)).thenApply(v -> v.getBoolValue());
}
@Override
- public CompletableFuture> mgetLongAsync(String... keys) {
- byte[][] bs = new byte[keys.length][];
- for (int i = 0; i < bs.length; i++) {
- bs[i] = keys[i].getBytes(StandardCharsets.UTF_8);
- }
- return sendAsync("MGET", keys[0], bs).thenApply(v -> {
- List list = (List) v.getListValue(keys[0], cryptor, long.class);
- Map map = new LinkedHashMap<>();
- for (int i = 0; i < keys.length; i++) {
- Object obj = list.get(i);
- if (obj != null) {
- map.put(keys[i], list.get(i));
- }
- }
- return map;
- });
+ public CompletableFuture> srandmemberAsync(String key, Type componentType, int count) {
+ return sendAsync(RedisCommand.SRANDMEMBER, key, keyArgs(key, count)).thenApply(v -> v.getListValue(key, cryptor, componentType));
}
@Override
- public CompletableFuture> mgetStringAsync(final String... keys) {
- byte[][] bs = new byte[keys.length][];
- for (int i = 0; i < bs.length; i++) {
- bs[i] = keys[i].getBytes(StandardCharsets.UTF_8);
- }
- return sendAsync("MGET", keys[0], bs).thenApply(v -> {
- List list = (List) v.getListValue(keys[0], cryptor, String.class);
- Map map = new LinkedHashMap<>();
- for (int i = 0; i < keys.length; i++) {
- Object obj = list.get(i);
- if (obj != null) {
- map.put(keys[i], list.get(i));
- }
- }
- return map;
- });
+ public CompletableFuture> sdiffAsync(final String key, final Type componentType, final String... key2s) {
+ return sendAsync(RedisCommand.SDIFF, key, keysArgs(key, key2s)).thenApply(v -> v.getSetValue(key, cryptor, componentType));
}
@Override
- public CompletableFuture> mgetAsync(final Type componentType, final String... keys) {
- byte[][] bs = new byte[keys.length][];
- for (int i = 0; i < bs.length; i++) {
- bs[i] = keys[i].getBytes(StandardCharsets.UTF_8);
- }
- return sendAsync("MGET", keys[0], bs).thenApply(v -> {
- List list = (List) v.getListValue(keys[0], cryptor, componentType);
- Map map = new LinkedHashMap<>();
- for (int i = 0; i < keys.length; i++) {
- Object obj = list.get(i);
- if (obj != null) {
- map.put(keys[i], list.get(i));
- }
- }
- return map;
- });
+ public CompletableFuture sdiffstoreAsync(final String key, final String srcKey, final String... srcKey2s) {
+ return sendAsync(RedisCommand.SDIFFSTORE, key, keysArgs(Utility.append(key, srcKey, srcKey2s))).thenApply(v -> v.getLongValue(0L));
}
@Override
- public CompletableFuture> mgetBytesAsync(final String... keys) {
- byte[][] bs = new byte[keys.length][];
- for (int i = 0; i < bs.length; i++) {
- bs[i] = keys[i].getBytes(StandardCharsets.UTF_8);
- }
- return sendAsync("MGET", keys[0], bs).thenApply(v -> {
- List list = (List) v.getListValue(keys[0], cryptor, byte[].class);
- Map map = new LinkedHashMap<>();
- for (int i = 0; i < keys.length; i++) {
- Object obj = list.get(i);
- if (obj != null) {
- map.put(keys[i], list.get(i));
- }
- }
- return map;
- });
+ public CompletableFuture> sinterAsync(final String key, final Type componentType, final String... key2s) {
+ return sendAsync(RedisCommand.SINTER, key, keysArgs(key, key2s)).thenApply(v -> v.getSetValue(key, cryptor, componentType));
+ }
+
+ @Override
+ public CompletableFuture sinterstoreAsync(final String key, final String srcKey, final String... srcKey2s) {
+ return sendAsync(RedisCommand.SINTERSTORE, key, keysArgs(Utility.append(key, srcKey, srcKey2s))).thenApply(v -> v.getLongValue(0L));
+ }
+
+ @Override
+ public CompletableFuture> sunionAsync(final String key, final Type componentType, final String... key2s) {
+ return sendAsync(RedisCommand.SUNION, key, keysArgs(key, key2s)).thenApply(v -> v.getSetValue(key, cryptor, componentType));
+ }
+
+ @Override
+ public CompletableFuture sunionstoreAsync(final String key, final String srcKey, final String... srcKey2s) {
+ return sendAsync(RedisCommand.SUNIONSTORE, key, keysArgs(Utility.append(key, srcKey, srcKey2s))).thenApply(v -> v.getLongValue(0L));
+ }
+
+ @Override
+ public CompletableFuture> mgetAsync(final Type componentType, final String... keys) {
+ return sendAsync(RedisCommand.MGET, keys[0], keysArgs(keys)).thenApply(v -> (List) v.getListValue(keys[0], cryptor, componentType));
}
@Override
public CompletableFuture>> smembersAsync(final Type componentType, final String... keys) {
- final CompletableFuture>> rsFuture = new CompletableFuture<>();
- final Map> map = new LinkedHashMap<>();
- final ReentrantLock mapLock = new ReentrantLock();
- final CompletableFuture[] futures = new CompletableFuture[keys.length];
+ final RedisCacheRequest[] requests = new RedisCacheRequest[keys.length];
for (int i = 0; i < keys.length; i++) {
- final String key = keys[i];
- futures[i] = sendAsync("SMEMBERS", key, keySetArgs(key)).thenAccept(v -> {
- Set c = v.getSetValue(key, cryptor, componentType);
- if (c != null) {
- mapLock.lock();
- try {
- map.put(key, c);
- } finally {
- mapLock.unlock();
- }
- }
- });
+ String key = keys[i];
+ requests[i] = RedisCacheRequest.create(RedisCommand.SMEMBERS, key, keyArgs(key));
}
- CompletableFuture.allOf(futures).whenComplete((w, e) -> {
- if (e != null) {
- rsFuture.completeExceptionally(e);
- } else {
- rsFuture.complete(map);
+ return sendAsync(requests).thenApply(list -> {
+ final Map> map = new LinkedHashMap<>();
+ for (int i = 0; i < keys.length; i++) {
+ String key = keys[i];
+ Set c = list.get(i).getSetValue(key, cryptor, componentType);
+ if (c != null) {
+ map.put(key, c);
+ }
}
+ return map;
});
- return rsFuture;
}
@Override
- public CompletableFuture>> lrangeAsync(final Type componentType, final String... keys) {
- final CompletableFuture>> rsFuture = new CompletableFuture<>();
- final Map> map = new LinkedHashMap<>();
- final ReentrantLock mapLock = new ReentrantLock();
- final CompletableFuture[] futures = new CompletableFuture[keys.length];
+ public