关于redis:聊聊claudb的importRDB

序

本文次要钻研一下 claudb 的 importRDB

importRDB

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/DBServerState.java

public class DBServerState {

    //......

  public void importRDB(InputStream input) throws IOException {RDBInputStream rdb = new RDBInputStream(input);

    Map<Integer, Map<DatabaseKey, DatabaseValue>> load = rdb.parse();
    for (Map.Entry<Integer, Map<DatabaseKey, DatabaseValue>> entry : load.entrySet()) {databases.get(entry.getKey()).overrideAll(ImmutableMap.from(entry.getValue()));
    }
  }

    //......
}

• importRDB 办法创立 RDBInputStream，而后执行其 parse 办法进行解析，之后遍历解析后果，挨个执行 databases.get(entry.getKey()).overrideAll(ImmutableMap.from(entry.getValue()))

RDBInputStream

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/persistence/RDBInputStream.java

public class RDBInputStream {private static final SafeString REDIS_PREAMBLE = safeString("REDIS");

  private static final long TO_MILLIS = 1000L;

  private static final int HASH = 0x04;
  private static final int SORTED_SET = 0x03;
  private static final int SET = 0x02;
  private static final int LIST = 0x01;
  private static final int STRING = 0x00;

  private static final int TTL_MILLISECONDS = 0xFC;
  private static final int TTL_SECONDS = 0xFD;
  private static final int SELECT = 0xFE;
  private static final int END_OF_STREAM = 0xFF;

  private static final int REDIS_VERSION = 6;
  private static final int VERSION_LENGTH = 4;
  private static final int REDIS_LENGTH = 5;

  private final CheckedInputStream in;

  public RDBInputStream(InputStream in) {this.in = new CheckedInputStream(in, new CRC64());
  }

  public Map<Integer, Map<DatabaseKey, DatabaseValue>> parse() throws IOException {Map<Integer, Map<DatabaseKey, DatabaseValue>> databases = new HashMap<>();

    int version = version();

    if (version > REDIS_VERSION) {throw new IOException("invalid version:" + version);
    }

    Long expireTime = null;
    HashMap<DatabaseKey, DatabaseValue> db = null;
    for (boolean end = false; !end;) {int read = in.read();
      switch (read) {
      case SELECT:
        db = new HashMap<>();
        databases.put(readLength(), db);
        break;
      case TTL_SECONDS:
        expireTime = parseTimeSeconds();
        break;
      case TTL_MILLISECONDS:
        expireTime = parseTimeMillis();
        break;
      case STRING:
        ensure(db, readKey(), readString(expireTime));
        expireTime = null;
        break;
      case LIST:
        ensure(db, readKey(), readList(expireTime));
        expireTime = null;
        break;
      case SET:
        ensure(db, readKey(), readSet(expireTime));
        expireTime = null;
        break;
      case SORTED_SET:
        ensure(db, readKey(), readSortedSet(expireTime));
        expireTime = null;
        break;
      case HASH:
        ensure(db, readKey(), readHash(expireTime));
        expireTime = null;
        break;
      case END_OF_STREAM:
        // end of stream
        end = true;
        db = null;
        expireTime = null;
        break;
      default:
        throw new IOException("not supported:" + read);
      }
    }

    verifyChecksum();

    return databases;
  }

  private int version() throws IOException {SafeString redis = new SafeString(read(REDIS_LENGTH));
    if (!redis.equals(REDIS_PREAMBLE)) {throw new IOException("not valid stream");
    }
    return parseVersion(read(VERSION_LENGTH));
  }

  private int parseVersion(byte[] version) {StringBuilder sb = new StringBuilder();
    for (byte b : version) {sb.append((char) b);
    }
    return Integer.parseInt(sb.toString());
  }

  private void verifyChecksum() throws IOException {long calculated = in.getChecksum().getValue();

    long readed = parseChecksum();

    if (calculated != readed) {throw new IOException("invalid checksum:" + readed);
    }
  }

  private long parseChecksum() throws IOException {return ByteUtils.byteArrayToLong(read(Long.BYTES));
  }

  //......

}

• RDBInputStream 的结构器应用 CheckedInputStream 对 InputStream 进行包装；其 parse 办法先解析 version 进行校验，之后循环进行 in.read()，别离解析 SELECT、TTL_SECONDS、TTL_MILLISECONDS、STRING、LIST、SET、SORTED_SET、HASH、END_OF_STREAM；最初执行 verifyChecksum 校验 checksum

readKey

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/persistence/RDBInputStream.java

  private DatabaseKey readKey() throws IOException {return new DatabaseKey(readSafeString());
  }

  private SafeString readSafeString() throws IOException {int length = readLength();
    return new SafeString(read(length));
  }

  private int readLength() throws IOException {int length = in.read();
    if (length < 0x40) {
      // 1 byte: 00XXXXXX
      return length;
    } else if (length < 0x80) {
      // 2 bytes: 01XXXXXX XXXXXXXX
      int next = in.read();
      return readLength(length, next);
    } else {
      // 5 bytes: 10...... XXXXXXXX XXXXXXXX XXXXXXXX XXXXXXXX
      return byteArrayToInt(read(Integer.BYTES));
    }
  }

  private byte[] read(int size) throws IOException {byte[] array = new byte[size];
    int read = in.read(array);
    if (read != size) {throw new IOException("error reading stream");
    }
    return array;
  }      

• readKey 办法执行 readSafeString，先 readLength，在依据 length 进行 read

readString

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/persistence/RDBInputStream.java

  private DatabaseValue readString(Long expireTime) throws IOException {return string(readSafeString()).expiredAt(expireTime != null ? ofEpochMilli(expireTime) : null);
  }

• readString 办法先 readSafeString，而后转为 DatabaseValue 再设置 expiredAt

readList

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/persistence/RDBInputStream.java

  private DatabaseValue readList(Long expireTime) throws IOException {int size = readLength();
    List<SafeString> list = new LinkedList<>();
    for (int i = 0; i < size; i++) {list.add(readSafeString());
    }
    return list(list).expiredAt(expireTime != null ? ofEpochMilli(expireTime) : null);
  }

• readList 办法先通过 readLength 读取 list 元素个数，在挨个执行 readSafeString，最初将 list 转为 DatabaseValue 再设置 expiredAt

readSet

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/persistence/RDBInputStream.java

  private DatabaseValue readSet(Long expireTime) throws IOException {int size = readLength();
    Set<SafeString> set = new LinkedHashSet<>();
    for (int i = 0; i < size; i++) {set.add(readSafeString());
    }
    return set(set).expiredAt(expireTime != null ? ofEpochMilli(expireTime) : null);
  }

• readSet 办法先通过 readLength 读取 set 元素个数，在挨个执行 readSafeString，最初将 set 转为 DatabaseValue 再设置 expiredAt

readSortedSet

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/persistence/RDBInputStream.java

  private DatabaseValue readSortedSet(Long expireTime) throws IOException {int size = readLength();
    Set<Entry<Double, SafeString>> entries = new LinkedHashSet<>();
    for (int i = 0; i < size; i++) {SafeString value = readSafeString();
      Double score = readDouble();
      entries.add(score(score, value));
    }
    return zset(entries).expiredAt(expireTime != null ? ofEpochMilli(expireTime) : null);
  }

• readSortedSet 办法先通过 readLength 读取 set 元素个数，在挨个执行 readSafeString、readDouble，增加到 LinkedHashSet，最初将 entries 转为 DatabaseValue 再设置 expiredAt；与 readSet 的差异在于多读取了 score

readHash

  private DatabaseValue readHash(Long expireTime) throws IOException {int size = readLength();
    Set<Tuple2<SafeString, SafeString>> entries = new LinkedHashSet<>();
    for (int i = 0; i < size; i++) {entries.add(entry(readSafeString(), readSafeString()));
    }
    return hash(entries).expiredAt(expireTime != null ? ofEpochMilli(expireTime) : null);
  }

• readHash 办法先通过 readLength 读取 hash 元素个数，在挨个执行 readSafeString、readSafeString，转为 entry 增加到 LinkedHashSet，最初将 entries 转为 DatabaseValue 再设置 expiredAt

小结

importRDB 办法创立 RDBInputStream，而后执行其 parse 办法进行解析，之后遍历解析后果，挨个执行 databases.get(entry.getKey()).overrideAll(ImmutableMap.from(entry.getValue()))

doc

• DBServerState