关于redis:聊聊claudb的exportRDB

本文次要钻研一下 claudb 的 exportRDB

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

public class DBServerState {

  //......

  public void exportRDB(OutputStream output) throws IOException {RDBOutputStream rdb = new RDBOutputStream(output);
    rdb.preamble(RDB_VERSION);
    for (int i = 0; i < databases.size(); i++) {Database db = databases.get(i);
      if (!db.isEmpty()) {rdb.select(i);
        rdb.dabatase(db);
      }
    }
    rdb.end();}

  //......

}

exportRDB 办法先通过 rdb.preamble(RDB_VERSION) 写入 redis 魔数及版本；而后遍历 databases，挨个执行 rdb.select(i) 写入 SELECT 及 db 的长度，在执行 rdb.dabatase(db)，遍历 entry，挨个按 expiredAt、type、key、value 写入数据；end 办法写入 END_OF_STREAM，而后再写入 checksum

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

public class RDBOutputStream {private static final byte[] REDIS = safeString("REDIS").getBytes();

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

  private final CheckedOutputStream out;

  public RDBOutputStream(OutputStream out) {super();
    this.out = new CheckedOutputStream(out, new CRC64());
  }

  public void preamble(int version) throws IOException {out.write(REDIS);
    out.write(version(version));
  }

  private byte[] version(int version) {StringBuilder sb = new StringBuilder(String.valueOf(version));
    for (int i = sb.length(); i < Integer.BYTES; i++) {sb.insert(0, '0');
    }
    return sb.toString().getBytes(StandardCharsets.UTF_8);
  }

  public void select(int db) throws IOException {out.write(SELECT);
    length(db);
  }

  public void dabatase(Database db) throws IOException {for (Tuple2<DatabaseKey, DatabaseValue> entry : db.entrySet()) {value(entry.get1(), entry.get2());
    }
  }

  private void value(DatabaseKey key, DatabaseValue value) throws IOException {expiredAt(value.getExpiredAt());
    type(value.getType());
    key(key);
    value(value);
  }

  private void expiredAt(Instant expiredAt) throws IOException {if (expiredAt != null) {out.write(TTL_MILLISECONDS);
      out.write(ByteUtils.toByteArray(expiredAt.toEpochMilli()));
    }
  }

  private void type(DataType type) throws IOException {out.write(type.ordinal());
  }

  private void key(DatabaseKey key) throws IOException {string(key.getValue());
  }

  private void value(DatabaseValue value) throws IOException {switch (value.getType()) {
    case STRING:
      string(value.getString());
      break;
    case LIST:
      list(value.getList());
      break;
    case HASH:
      hash(value.getHash());
      break;
    case SET:
      set(value.getSet());
      break;
    case ZSET:
      zset(value.getSortedSet());
      break;
    default:
      break;
    }
  }

  private void length(int length) throws IOException {if (length < 0x40) {
      // 1 byte: 00XXXXXX
      out.write(length);
    } else if (length < 0x4000) {
      // 2 bytes: 01XXXXXX XXXXXXXX
      int b1 = length >> 8;
      int b2 = length & 0xFF;
      out.write(0x40 | b1);
      out.write(b2);
    } else {
      // 5 bytes: 10...... XXXXXXXX XXXXXXXX XXXXXXXX XXXXXXXX
      out.write(0x80);
      out.write(toByteArray(length));
    }
  }

  private void string(String value) throws IOException {string(safeString(value));
  }

  private void string(SafeString value) throws IOException {byte[] bytes = value.getBytes();
    length(bytes.length);
    out.write(bytes);
  }

  private void string(double value) throws IOException {string(String.valueOf(value));
  }

  private void list(ImmutableList<SafeString> value) throws IOException {length(value.size());
    for (SafeString item : value) {string(item);
    }
  }

  private void hash(ImmutableMap<SafeString, SafeString> value) throws IOException {length(value.size());
    for (Tuple2<SafeString, SafeString> entry : value.entries()) {string(entry.get1());
      string(entry.get2());
    }
  }

  private void set(ImmutableSet<SafeString> value) throws IOException {length(value.size());
    for (SafeString item : value) {string(item);
    }
  }

  private void zset(NavigableSet<Entry<Double, SafeString>> value) throws IOException {length(value.size());
    for (Entry<Double, SafeString> item : value) {string(item.getValue());
      string(item.getKey());
    }
  }

  public void end() throws IOException {out.write(END_OF_STREAM);
    out.write(toByteArray(out.getChecksum().getValue()));
    out.flush();}
}

RDBOutputStream 的结构器用 CheckedOutputStream 包装了 OutputStream；其 dabatase 办法遍历 db.entrySet()，挨个执行 value 办法；value 办法先别离执行 expiredAt、type、key、value 办法；value 办法针对 STRING、LIST、HASH、SET、ZSET 这几种 value 类型做了不同的解决；string 办法间接写入 string；list 办法先写入 list 大小，再挨个写入 list 元素；hash 办法先写入 hash 大小，再挨个写入 key 和 value；set 先写入 set 大小，在挨个写入 set 元素；zset 先写入 zset 大小，再挨个写入 value 和 score；end 办法写入 END_OF_STREAM，而后再写入 checksum

java.base/java/util/zip/CheckedOutputStream.java

public
class CheckedOutputStream extends FilterOutputStream {
    private Checksum cksum;

    /**
     * Creates an output stream with the specified Checksum.
     * @param out the output stream
     * @param cksum the checksum
     */
    public CheckedOutputStream(OutputStream out, Checksum cksum) {super(out);
        this.cksum = cksum;
    }

    /**
     * Writes a byte. Will block until the byte is actually written.
     * @param b the byte to be written
     * @exception IOException if an I/O error has occurred
     */
    public void write(int b) throws IOException {out.write(b);
        cksum.update(b);
    }

    /**
     * Writes an array of bytes. Will block until the bytes are
     * actually written.
     * @param b the data to be written
     * @param off the start offset of the data
     * @param len the number of bytes to be written
     * @exception IOException if an I/O error has occurred
     */
    public void write(byte[] b, int off, int len) throws IOException {out.write(b, off, len);
        cksum.update(b, off, len);
    }

    /**
     * Returns the Checksum for this output stream.
     * @return the Checksum
     */
    public Checksum getChecksum() {return cksum;}
}

CheckedOutputStream 继承了 FilterOutputStream，其结构器要求输出 OutputStream 及 Checksum，每次 write 的时候都会执行 cksum.update；其 getChecksum 办法间接返回 cksum

public class CRC64 implements Checksum {

  private static final int LOOKUPTABLE_SIZE = 256;
  private static final long POLY64REV = 0xC96C5795D7870F42L;
  private static final long LOOKUPTABLE[] = new long[LOOKUPTABLE_SIZE];

  private long crc = -1;

  static {for (int b = 0; b < LOOKUPTABLE.length; ++b) {
      long r = b;
      for (int i = 0; i < Long.BYTES; ++i) {if ((r & 1) == 1) {r = (r >>> 1) ^ POLY64REV;
        } else {r >>>= 1;}
      }
      LOOKUPTABLE[b] = r;
    }
  }

  @Override
  public void update(int b) {crc = LOOKUPTABLE[((b & 0xFF) ^ (int) crc) & 0xFF] ^ (crc >>> 8);
  }

  @Override
  public void update(byte[] buf, int off, int len) {
    int end = off + len;

    while (off < end) {crc = LOOKUPTABLE[(buf[off++] ^ (int) crc) & 0xFF] ^ (crc >>> 8);
    }
  }

  @Override
  public long getValue() {return ~crc;}

  @Override
  public void reset() {crc = -1;}
}

CRC64 实现了 Checksum 接口，其 update 办法会借用 LOOKUPTABLE 更新 crc

exportRDB 办法先通过 rdb.preamble(RDB_VERSION) 写入 redis 魔数及版本；而后遍历 databases，挨个执行 rdb.select(i) 写入 SELECT 及 db 的长度，在执行 rdb.dabatase(db)，遍历 entry，挨个按 expiredAt、type、key、value 写入数据；end 办法写入 END_OF_STREAM，而后再写入 checksum

DBServerState

关于redis:聊聊claudb的exportRDB

序

exportRDB

RDBOutputStream

CheckedOutputStream

CRC64

小结

doc