Skip to content
Open
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -40,15 +40,24 @@ public enum RedisOpType {
LREM(false, 4),
LSET(false, 4),
LTRIM(false, 4),
BLMPOP(true,-5),
LMPOP(true,-4),

// Hash single
HDEL(false, -3),
HINCRBY(false, 4),
HINCRBYFLOAT(false, 4),
HMSET(false, -4),
HSET(false, -4),
HSETEX(false, -4),
HSETEX(false, -6),
HSETNX(false, 4),
HEXPIREAT(false,-6),
HPEXPIREAT(false,-6),
HEXPIRE(false,-6),
HPEXPIRE(false,-6),
HGETDEL(false,-5),
HPERSIST(false,-5),
HGETEX(false,-5),

// Set single
SADD(false, -3),
Expand All @@ -62,13 +71,18 @@ public enum RedisOpType {
ZREMRANGEBYLEX(false, 4),
ZREMRANGEBYRANK(false, 4),
ZREMRANGEBYSCORE(false, 4),
ZMPOP(true,-4),
BZMPOP(true,-5),

// Stream single
XADD(false, -5),
XDEL(false, -3),
XSETID(false, 3),
XGROUP(false, -2),
XCLAIM(false, -6),
XDELEX(false,-6),
XACKDEL(false,-7),


// TTL single
EXPIRE(false, 3),
Expand All @@ -83,6 +97,8 @@ public enum RedisOpType {

// Bit single
SETBIT(false, 4),
BITCOUNT(false, 4),


// String multi
DEL(true, -2),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ private RedisOpItemParser() {
public static RedisOpItem parse(RedisOpParser redisOpParser, Object[] payload) {
RedisOpItem redisOpItem = new RedisOpItem();
try {
RedisOp redisOp = redisOpParser.parse(payload);
RedisOp redisOp = redisOpParser.parse(payload);
redisOpItem.setRedisOpType(redisOp.getOpType());
redisOpItem.setGtid(redisOp.getOpGtid());
redisOpItem.setDbId(redisOp.getDbId());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,10 @@ public enum RedisOpMultiKeysEnum {
MSETNX(RedisOpType.MSETNX, 1, 2),
DEL(RedisOpType.DEL, 1, 1),
UNLINK(RedisOpType.UNLINK, 1, 1),
LMPOP(RedisOpType.LMPOP, 2, 1),
BLMPOP(RedisOpType.BLMPOP, 3, 1),
ZMPOP(RedisOpType.ZMPOP, 2, 1),
BZMPOP(RedisOpType.BZMPOP, 3, 1),

//crdt,
CRDT_DEL_REG(RedisOpType.CRDT_DEL_REG, 1, 1),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,10 @@
import com.ctrip.xpipe.redis.core.redis.operation.RedisOpType;
import com.ctrip.xpipe.redis.core.redis.operation.op.RedisOpMultiKVs;
import com.ctrip.xpipe.tuple.Pair;
import org.checkerframework.checker.units.qual.A;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
Expand All @@ -17,6 +19,12 @@
*/
public class RedisOpMultiKeysParser extends AbstractRedisOpParser implements RedisOpParser {

private static final byte[] MIN_BYTES = "MIN".getBytes();
private static final byte[] MIN_BYTES_LOWER = "min".getBytes();

private static final byte[] MAX_BYTES = "MAX".getBytes();
private static final byte[] MAX_BYTES_LOWER = "max".getBytes();

private RedisOpType redisOpType;
private Integer keyStartIndex;
private Integer kvNum;
Expand All @@ -37,7 +45,17 @@ public RedisOp parse(byte[][] args) {
List<Pair<RedisKey, byte[]>> kvs = new ArrayList<>();

int i = keyStartIndex;
while (i < args.length) {
int argLen = args.length;

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

扫一下8.0注册了getkeyrequests_proc的命令,如果都符合可以用这种写法;否则可以参考下Redis的写法,对于特殊的key获取定义方法注册到命令定义枚举

if(keyStartIndex>1){
int numKeys;
try {
numKeys = Integer.parseInt(new String(args[keyStartIndex-1]));
} catch (NumberFormatException e) {
throw new IllegalArgumentException("Invalid numkeys for " + redisOpType);
}
argLen = i+numKeys;
}
while (i < argLen) {
RedisKey key = new RedisKey(args[i++]);
kvs.add(kvNum == 1 ? Pair.of(key, null) : Pair.of(key, args[i++]));
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,14 @@ public enum RedisOpWithSubKeysEnum {
HINCRBYFLOAT(RedisOpType.HINCRBYFLOAT, 1, 2,false),
HMSET(RedisOpType.HMSET, 1, 2,false),
HSET(RedisOpType.HSET, 1, 2,false),
HSETEX(RedisOpType.HSET, 1, 2,false),
HSETEX(RedisOpType.HSETEX, 1, 2,false),
HGETEX(RedisOpType.HGETEX, 1, 1,false),
HGETDEL(RedisOpType.HGETDEL, 1, 1,false),
HEXPIRE(RedisOpType.HEXPIRE, 1, 1,false),
HEXPIREAT(RedisOpType.HEXPIREAT, 1, 1,false),
HPEXPIRE(RedisOpType.HPEXPIRE, 1, 1,false),
HPEXPIREAT(RedisOpType.HPEXPIREAT, 1, 1,false),
HPERSIST(RedisOpType.HPERSIST, 1, 1,false),
HSETNX(RedisOpType.HSETNX, 1, 2,false),

ZADD(RedisOpType.ZADD, 1, 2,true),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,11 @@
import com.ctrip.xpipe.redis.core.redis.operation.RedisOp;
import com.ctrip.xpipe.redis.core.redis.operation.RedisOpParser;
import com.ctrip.xpipe.redis.core.redis.operation.RedisOpType;
import com.ctrip.xpipe.redis.core.redis.operation.op.RedisOpMultiKVs;
import com.ctrip.xpipe.redis.core.redis.operation.op.RedisOpMultiSubKey;
import com.ctrip.xpipe.tuple.Pair;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.nio.ByteBuffer;
import java.util.*;

/**
* @author tb
Expand All @@ -24,12 +22,28 @@ public class RedisOpWithSubKeysParser extends AbstractRedisOpParser implements R
private Integer kvNum;
private boolean kvReverse;

private byte[] NX_BYTES = new byte[]{'N','X'};
private byte[] XX_BYTES = new byte[]{'X','X'};
private byte[] GT_BYTES = new byte[]{'G','T'};
private byte[] LT_BYTES = new byte[]{'L','T'};
private byte[] CH_BYTES = new byte[]{'C','H'};
private byte[] INCR_BYTES = new byte[]{'I','N','C','R'};
private static final Set<ByteBuffer> NON_KEY_COMMANDS = new HashSet<>();
private static final Set<ByteBuffer> NON_KEY_COMMANDS_WITH_ARGS = new HashSet<>();

private static final byte[] FIELDS_BYTES = "FIELDS".getBytes();
private static final byte[] FIELDS_BYTES_LOWER = "fields".getBytes();

static {
// 无参数关键字
String[] noArgCmds = {
"NX", "nx", "XX", "xx", "GT", "gt", "LT", "lt", "CH", "ch",
"INCR", "incr", "KEEPTTL", "keepttl", "FNX", "fnx", "FXX", "fxx", "PERSIST", "persist"
};
for (String cmd : noArgCmds) {
NON_KEY_COMMANDS.add(ByteBuffer.wrap(cmd.getBytes()));
}
// 带一个参数的关键字
String[] oneArgCmds = { "EX", "ex", "PX", "px", "EXAT", "exat", "PXAT", "pxat" };
for (String cmd : oneArgCmds) {
NON_KEY_COMMANDS_WITH_ARGS.add(ByteBuffer.wrap(cmd.getBytes()));
NON_KEY_COMMANDS.add(ByteBuffer.wrap(cmd.getBytes())); // 也在无参数集合中,因为需要先识别
}
}

public RedisOpWithSubKeysParser(RedisOpType redisOpType, int keyStartIndex, int kvNum,boolean kvReverse) {
this.redisOpType = redisOpType;
Expand All @@ -43,6 +57,13 @@ public RedisOp parse(byte[][] args) {
Pair<RedisOpType, byte[][]> pair = redisOpType.transfer(redisOpType, args);
args = pair.getValue();

if(usesFieldsSyntax(redisOpType)) {
return parseFieldsCommand(redisOpType,args);
}
return parseGeneralCommands(args, pair);
}

private RedisOpMultiSubKey parseGeneralCommands(byte[][] args, Pair<RedisOpType, byte[][]> pair) {
// 计算容量 - 主键 + 子键数量
int subKeyCount = (args.length - keyStartIndex-1) / kvNum;
int capacity = 1 + subKeyCount;
Expand Down Expand Up @@ -76,7 +97,7 @@ public RedisOp parse(byte[][] args) {
subKeys.add(subKey);
}

return new RedisOpMultiSubKey(pair.getKey(), args, key,subKeys);
return new RedisOpMultiSubKey(pair.getKey(), args, key, subKeys);
}

@Override
Expand All @@ -85,8 +106,45 @@ public int getOrder() {
}

private boolean nonKey(byte[] args){
return Arrays.equals(NX_BYTES,args) || Arrays.equals(XX_BYTES,args)
|| Arrays.equals(GT_BYTES,args) || Arrays.equals(LT_BYTES,args)
|| Arrays.equals(CH_BYTES,args) || Arrays.equals(INCR_BYTES,args);
return NON_KEY_COMMANDS.contains(ByteBuffer.wrap(args));
}

private boolean usesFieldsSyntax(RedisOpType opType) {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

放到OpType定义里?

return opType == RedisOpType.HSETEX || opType == RedisOpType.HEXPIRE
|| opType == RedisOpType.HEXPIREAT || opType == RedisOpType.HGETEX
|| opType == RedisOpType.HGETDEL || opType == RedisOpType.HPEXPIRE
|| opType == RedisOpType.HPEXPIREAT || opType == RedisOpType.HPERSIST;
}

private RedisOp parseFieldsCommand(RedisOpType opType, byte[][] args) {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

和parseGeneralCommands的关系?

int idx = keyStartIndex;
RedisKey key = new RedisKey(args[idx++]);

// 跳过 FIELDS 之前的所有可选标志
while (idx < args.length) {
ByteBuffer tokenBuf = ByteBuffer.wrap(args[idx]);
if (NON_KEY_COMMANDS.contains(tokenBuf)) {
idx++;
if (NON_KEY_COMMANDS_WITH_ARGS.contains(tokenBuf) && idx < args.length) {
idx++; // 跳过标志的参数
}
} else if (Arrays.equals(args[idx], FIELDS_BYTES) || Arrays.equals(args[idx], FIELDS_BYTES_LOWER)) {
idx++;
break;
} else {
idx++;
}
}

if (idx >= args.length) throw new IllegalArgumentException("Missing numfields");
int numFields = Integer.parseInt(new String(args[idx++]));
List<RedisKey> subKeys = new ArrayList<>(numFields);

for (int i = 0; i < numFields; i++) {
if (idx >= args.length) throw new IllegalArgumentException("Incomplete field list");
subKeys.add(new RedisKey(args[idx]));
idx += kvNum;
}
return new RedisOpMultiSubKey(opType, args, key, subKeys);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,8 @@ private RdbConstant() {
public static final short REDIS_RDB_TYPE_HASH_ZIPLIST = 13;
public static final short REDIS_RDB_TYPE_LIST_QUICKLIST = 14;
public static final short REDIS_RDB_TYPE_STREAM_LISTPACKS = 15;
// public static final short REDIS_RDB_TYPE_BITMAP = 16;
public static final short REDIS_RDB_TYPE_BITMAP = 192;
public static final short REDIS_RDB_TYPE_BITMAP_9 = 16;
public static final short REDIS_RDB_TYPE_CRDT = 200;

// Redis 8.x 新增操作码 (rdb.h)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,8 @@ enum RdbType {
HASH_ZIPLIST(RdbConstant.REDIS_RDB_TYPE_HASH_ZIPLIST, false, RdbHashZipListParser::new),
LIST_QUICKLIST(RdbConstant.REDIS_RDB_TYPE_LIST_QUICKLIST, false, RdbQuickListParser::new),
STREAM_LISTPACKS(RdbConstant.REDIS_RDB_TYPE_STREAM_LISTPACKS, false, RdbStreamListpacksParser::new),
// BITMAP(RdbConstant.REDIS_RDB_TYPE_BITMAP, false, RdbBitmapParser::new),
BITMAP(RdbConstant.REDIS_RDB_TYPE_BITMAP, false, RdbBitmapParser::new),
BITMAP_9(9,RdbConstant.REDIS_RDB_TYPE_BITMAP_9, false, RdbBitmapParser::new),
CRDT(RdbConstant.REDIS_RDB_TYPE_CRDT, false, DefaultRdbCrdtParser::new),
// MODULE_AUX(RdbConstant.REDIS_RDB_OP_CODE_MODULE_AUX),
IDLE(RdbConstant.REDIS_RDB_OP_CODE_IDLE, true, RdbIdleParser::new),
Expand Down Expand Up @@ -132,15 +133,25 @@ enum RdbType {
STREAM_LISTPACKS_3(RdbConstant.REDIS_RDB_TYPE_STREAM_LISTPACKS_3, false, RdbStreamListpacks3Parser::new),
STREAM_LISTPACKS_4(RdbConstant.REDIS_RDB_TYPE_STREAM_LISTPACKS_4, false, RdbStreamListpacks4Parser::new);

private int version;
private short code;

private boolean rdbOp;

private Function<RdbParseContext, RdbParser> parserConstructor;

private static final Map<Short, RdbType> types = new HashMap<>();
private static final Map<Integer,Map<Short, RdbType>> versionTypes = new HashMap<>();

RdbType(short code, boolean rdbOp, Function<RdbParseContext, RdbParser> parserConstructor) {
this.version = 0;

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

复用RdbType(int version,short code, boolean rdbOp, Function<RdbParseContext, RdbParser> parserConstructor)?

this.code = code;
this.rdbOp = rdbOp;
this.parserConstructor = parserConstructor;
}

RdbType(int version,short code, boolean rdbOp, Function<RdbParseContext, RdbParser> parserConstructor) {
this.version = version;
this.code = code;
this.rdbOp = rdbOp;
this.parserConstructor = parserConstructor;
Expand All @@ -163,14 +174,28 @@ public RdbParser<?> makeParser(RdbParseContext parserManager) {

static {
for (RdbType rdbType : values()) {
types.put(rdbType.code, rdbType);
if(rdbType.version != 0) {
Map<Short, RdbType> versionMap = versionTypes.computeIfAbsent(rdbType.version, (version) -> new HashMap<>());
versionMap.put(rdbType.code, rdbType);
}else {
types.put(rdbType.code, rdbType);
}
}
}

public static RdbType findByCode(short code) {
return types.get(code);
}

public static RdbType findByCode(int version,short code) {
Map<Short,RdbType> versionMap = versionTypes.getOrDefault(version,types);
RdbType rdbType = versionMap.get(code);
if(rdbType == null){
return findByCode(code);
}
return rdbType;
}

}

String cSet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ public Void read(ByteBuf byteBuf) {

case READ_TYPE:
short type = byteBuf.readUnsignedByte();
RdbParseContext.RdbType newType = RdbParseContext.RdbType.findByCode(type);
RdbParseContext.RdbType newType = RdbParseContext.RdbType.findByCode(rdbVersion,type);
if (currentType == RdbParseContext.RdbType.AUX && newType != RdbParseContext.RdbType.AUX) {
auxFinished.set(true);
notifyAuxEnd(rdbParseContext.getAllAux());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,11 @@ private void propagateCmdIfNeed(byte[] value) {
RedisOpType.SET,
new byte[][]{RedisOpType.SET.name().getBytes(), context.getKey().get(), value},
context.getKey(), value));

notifyRedisOp(new RedisOpSingleKey(
RedisOpType.BITCOUNT,
new byte[][]{RedisOpType.BITCOUNT.name().getBytes(), context.getKey().get()},
context.getKey(), value));
}

@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -63,8 +63,8 @@ public Integer read(ByteBuf byteBuf) {
break;
case READ_DATA:
byte[] data = rdbStringParser.read(byteBuf);
rdbStringParser.reset();
if (data != null) {
rdbStringParser.reset();
readCnt++;

if (readCnt >= len.getLenValue()) {
Expand Down
Loading
Loading