关于redis:聊聊claudb的Database

序

本文次要钻研一下 claudb 的 Database

Database

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/data/Database.java

public interface Database {int size();

  boolean isEmpty();

  boolean containsKey(DatabaseKey key);

  DatabaseValue get(DatabaseKey key);

  DatabaseValue put(DatabaseKey key, DatabaseValue value);

  DatabaseValue remove(DatabaseKey key);

  void clear();

  ImmutableSet<DatabaseKey> keySet();

  Sequence<DatabaseValue> values();

  ImmutableSet<Tuple2<DatabaseKey, DatabaseValue>> entrySet();

  default SafeString getString(SafeString key) {return getOrDefault(safeKey(key), DatabaseValue.EMPTY_STRING).getString();}

  default ImmutableList<SafeString> getList(SafeString key) {return getOrDefault(safeKey(key), DatabaseValue.EMPTY_LIST).getList();}

  default ImmutableSet<SafeString> getSet(SafeString key) {return getOrDefault(safeKey(key), DatabaseValue.EMPTY_SET).getSet();}

  default NavigableSet<Entry<Double, SafeString>> getSortedSet(SafeString key) {return getOrDefault(safeKey(key), DatabaseValue.EMPTY_ZSET).getSortedSet();}

  default ImmutableMap<SafeString, SafeString> getHash(SafeString key) {return getOrDefault(safeKey(key), DatabaseValue.EMPTY_HASH).getHash();}

  default void putAll(ImmutableMap<? extends DatabaseKey, ? extends DatabaseValue> map) {map.forEach(this::put);
  }

  default DatabaseValue putIfAbsent(DatabaseKey key, DatabaseValue value) {DatabaseValue oldValue = get(key);
    if (oldValue == null) {oldValue = put(key, value);
    }
    return oldValue;
  }

  default DatabaseValue merge(DatabaseKey key, DatabaseValue value,
      BiFunction<DatabaseValue, DatabaseValue, DatabaseValue> remappingFunction) {DatabaseValue oldValue = get(key);
    DatabaseValue newValue = oldValue == null ? value : remappingFunction.apply(oldValue, value);
    if(newValue == null) {remove(key);
    } else {put(key, newValue);
    }
    return newValue;
  }

  default DatabaseValue getOrDefault(DatabaseKey key, DatabaseValue defaultValue) {DatabaseValue value = get(key);
    return (value != null || containsKey(key)) ? value : defaultValue;
  }

  default boolean isType(DatabaseKey key, DataType type) {DatabaseValue value = get(key);
    return value != null ? value.getType() == type : true;}

  default boolean rename(DatabaseKey from, DatabaseKey to) {DatabaseValue value = remove(from);
    if (value != null) {put(to, value);
      return true;
    }
    return false;
  }

  default void overrideAll(ImmutableMap<DatabaseKey, DatabaseValue> value) {clear();
    putAll(value);
  }

  default ImmutableSet<DatabaseKey> evictableKeys(Instant now) {return entrySet()
        .filter(entry -> entry.get2().isExpired(now))
        .map(Tuple2::get1);
  }
}

• Database 接口定义了 size、isEmpty、containsKey、get、put、remove、clear、keySet、values、entrySet 办法；同时还提供了 getString、getList、getSet、getSortedSet、getHash、putAll、putIfAbsent、merge、getOrDefault、isType、rename、overrideAll、evictableKeys 这几个 default 办法

OnHeapDatabase

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/data/OnHeapDatabase.java

public class OnHeapDatabase implements Database {

  private final Map<DatabaseKey, DatabaseValue> cache;

  public OnHeapDatabase(Map<DatabaseKey, DatabaseValue> cache) {this.cache = cache;}

  @Override
  public int size() {return cache.size();
  }

  @Override
  public boolean isEmpty() {return cache.isEmpty();
  }

  @Override
  public boolean containsKey(DatabaseKey key) {return cache.containsKey(key);
  }

  @Override
  public DatabaseValue get(DatabaseKey key) {DatabaseValue value = cache.get(key);
    if (value != null) {if (!value.isExpired(Instant.now())) {return value;}
      cache.remove(key);
    }
    return null;
  }

  @Override
  public DatabaseValue put(DatabaseKey key, DatabaseValue value) {DatabaseValue oldValue = cache.remove(key);
    cache.put(key, value);
    return oldValue;
  }

  @Override
  public DatabaseValue remove(DatabaseKey key) {return cache.remove(key);
  }

  @Override
  public void clear() {cache.clear();
  }

  @Override
  public ImmutableSet<DatabaseKey> keySet() {return ImmutableSet.from(cache.keySet());
  }

  @Override
  public Sequence<DatabaseValue> values() {return ImmutableSet.from(cache.values());
  }

  @Override
  public ImmutableSet<Tuple2<DatabaseKey, DatabaseValue>> entrySet() {return ImmutableSet.from(cache.entrySet()).map(Tuple::from);
  }
}

• OnHeapDatabase 实现了 Database 接口，它应用 Map 构造作为 cache；其 get 办法在取出 value 不为 null 时会判断该 value 是否过期，如果过期则移除该 key，返回 null；其 put 办法会先执行 remove 获取 oldValue，在 put 进去新值，最初返回 oldValue

OffHeapDatabase

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/data/OffHeapDatabase.java

public class OffHeapDatabase implements Database {

  private OHCache<DatabaseKey, DatabaseValue> cache;

  public OffHeapDatabase(OHCache<DatabaseKey, DatabaseValue> cache) {this.cache = cache;}

  @Override
  public int size() {return (int) cache.size();}

  @Override
  public boolean isEmpty() {return cache.size() == 0;
  }

  @Override
  public boolean containsKey(DatabaseKey key) {return cache.containsKey(key);
  }

  @Override
  public DatabaseValue get(DatabaseKey key) {DatabaseValue value = cache.get(key);
    if (value != null) {if (!value.isExpired(Instant.now())) {return value;}
      cache.remove(key);
    }
    return null;
  }

  @Override
  public DatabaseValue put(DatabaseKey key, DatabaseValue value) {cache.put(key, value);
    return value;
  }

  @Override
  public DatabaseValue remove(DatabaseKey key) {DatabaseValue value = get(key);
    cache.remove(key);
    return value;
  }

  @Override
  public void clear() {cache.clear();
  }

  @Override
  public ImmutableSet<DatabaseKey> keySet() {Set<DatabaseKey> keys = new HashSet<>();
    try (CloseableIterator<DatabaseKey> iterator = cache.keyIterator()) {while (iterator.hasNext()) {keys.add(iterator.next());
      }
    } catch(IOException e) {throw new UncheckedIOException(e);
    }
    return ImmutableSet.from(keys);
  }

  @Override
  public Sequence<DatabaseValue> values() {List<DatabaseValue> values = new LinkedList<>();
    for (DatabaseKey key : keySet()) {values.add(cache.get(key));
    }
    return ImmutableList.from(values);
  }

  @Override
  public ImmutableSet<Tuple2<DatabaseKey, DatabaseValue>> entrySet() {return keySet().map(key -> Tuple.of(key, get(key)));
  }
}

• OffHeapDatabase 实现了 Database 接口，它应用 OHCache 作为 cache，其 get 办法在取出 value 不为 null 时会判断该 value 是否过期，如果过期则移除该 key，返回 null；其 put 办法间接往 cache 笼罩该 key，返回的是新值

DatabaseFactory

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/data/DatabaseFactory.java

public interface DatabaseFactory {Database create(String name);
  void clear();}

• DatabaseFactory 接口定义了 create、clear 办法

OnHeapDatabaseFactory

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/data/OnHeapDatabaseFactory.java

public class OnHeapDatabaseFactory implements DatabaseFactory {

  @Override
  public Database create(String name) {return new OnHeapDatabase(new HashMap<>());
  }

  @Override
  public void clear() {// nothing to clear}
}

• OnHeapDatabaseFactory 实现了 DatabaseFactory 接口，其 create 应用 HashMap 创立 OnHeapDatabase

OffHeapDatabaseFactory

claudb-1.7.1/src/main/java/com/github/tonivade/claudb/data/OffHeapDatabaseFactory.java

public class OffHeapDatabaseFactory implements DatabaseFactory {

  @Override
  public Database create(String name) {return new OffHeapDatabase(createCache());
  }

  private OHCache<DatabaseKey, DatabaseValue> createCache() {return builder()
        .eviction(Eviction.NONE)
        .throwOOME(true)
        .keySerializer(new FSTSerializer<>())
        .valueSerializer(new FSTSerializer<>())
        .build();}

  private OHCacheBuilder<DatabaseKey, DatabaseValue> builder() {return OHCacheBuilder.newBuilder();
  }

  @Override
  public void clear() {// nothing to do}

  private static class FSTSerializer<E> implements CacheSerializer<E> {private static final FSTConfiguration FST = FSTConfiguration.createDefaultConfiguration();

    static {FST.registerClass(DatabaseValue.class);
      FST.registerClass(DatabaseKey.class);
      FST.registerClass(SafeString.class);
      FST.registerClass(SortedSet.class);
    }

    @Override
    public void serialize(E value, ByteBuffer buf) {byte[] array = FST.asByteArray(value);
      buf.putInt(array.length);
      buf.put(array);
    }

    @SuppressWarnings("unchecked")
    @Override
    public E deserialize(ByteBuffer buf) {int length = buf.getInt();
      byte[] array = new byte[length];
      buf.get(array);
      return (E) FST.asObject(array);
    }

    @Override
    public int serializedSize(E value) {return FST.asByteArray(value).length + Integer.BYTES;
    }
  }
}

• OffHeapDatabaseFactory 实现了 DatabaseFactory 接口，其 create 办法创立的是 OffHeapDatabase；其 createCache 办法应用 OHCacheBuilder 来结构 OHCache，其 eviction 为 NONE，其 throwOOME 为 true；其 keySerializer 及 valueSerializer 均为 FSTSerializer

小结

claudb 提供了 OnHeapDatabase、OffHeapDatabase 两种实现，前者应用 HashMap，后者应用 OHCache

doc

• Database