PalDB 读数据过程-阿里云开发者社区

开发者社区> 晴天哥> 正文

PalDB 读数据过程

简介: 开篇 PalDB 介绍 PalDB 写数据过程 PalDB 读数据过程 PalDB 线程安全版本 PalDB reader的多级缓存  在解释PalDB的读过程之前先解释下为啥PalDB读的性能比较理想,其实PalDB可以理解为有多级缓存: 一级缓存是StorageCache对象,里面是用LinkedHashMap实现的缓存 二级缓存是StorageReader对象,通过mmap实现的文件到内存的映射。
+关注继续查看

开篇


PalDB reader的多级缓存

 在解释PalDB的读过程之前先解释下为啥PalDB读的性能比较理想,其实PalDB可以理解为有多级缓存:

  • 一级缓存是StorageCache对象,里面是用LinkedHashMap实现的缓存
  • 二级缓存是StorageReader对象,通过mmap实现的文件到内存的映射。
public final class ReaderImpl implements StoreReader {
  // Logger
  private final static Logger LOGGER = Logger.getLogger(ReaderImpl.class.getName());
  // Configuration
  private final Configuration config;
  // Buffer
  private final DataInputOutput dataInputOutput = new DataInputOutput();
  // Storage
  private final StorageReader storage;
  // Serialization
  private final StorageSerialization serialization;
  // Cache
  private final StorageCache cache;
  // File
  private final File file;
  // Opened?
  private boolean opened;




private StorageCache(Configuration config) {
    cache = new LinkedHashMap(config.getInt(Configuration.CACHE_INITIAL_CAPACITY),
        config.getFloat(Configuration.CACHE_LOAD_FACTOR), true) {
      @Override
      protected boolean removeEldestEntry(Map.Entry eldest) {
        boolean res = currentWeight > maxWeight;
        if (res) {
          Object key = eldest.getKey();
          Object value = eldest.getValue();
          currentWeight -= getWeight(key) + getWeight(value) + OVERHEAD;
        }
        return res;
      }
    };


PalDB 的读取过程

PalDB宏观的读取过程

 PalDB的读取过程也非常简单明了,就是标准的带有缓存的读取过程:

  • 判断缓存是否有数据,有则直接从cache中取出返回
  • 通过storage.get从mmap的文件中读取数据
  • 将数据放到缓存后直接返回数据
public <K> K get(Object key, K defaultValue) {
    checkOpen();
    if (key == null) {
      throw new NullPointerException("The key can't be null");
    }
    K value = cache.get(key);
    if (value == null) {
      try {
        byte[] valueBytes = storage.get(serialization.serializeKey(key));
        if (valueBytes != null) {

          Object v = serialization.deserialize(dataInputOutput.reset(valueBytes));
          cache.put(key, v);
          return (K) v;
        } else {
          return defaultValue;
        }
      } catch (Exception ex) {
        throw new RuntimeException(ex);
      }
    } else if (value == StorageCache.NULL_VALUE) {
      return null;
    }
    return value;
  }


StorageReader的读取过程

 PalDB读取过程是一个根据key进行hash定位slot的过程,整个过程如下:

  • 对key进行hash的得到hash值,获取key下的slot的数量
  • 在key的mmap内容中定位到key在PalDB中的位置,得到对应的value的偏移量
  • 通过value的偏移量在value的mmap内容中定位到value对应的值并返回

 在读取value的mmap的过程中,会根据偏移量对mmap的文件个数求余后定位mmap的文件后进行读取。

public byte[] get(byte[] key)
      throws IOException {
    int keyLength = key.length;
    if (keyLength >= slots.length || keyCounts[keyLength] == 0) {
      return null;
    }
    long hash = (long) hashUtils.hash(key);
    int numSlots = slots[keyLength];
    int slotSize = slotSizes[keyLength];
    int indexOffset = indexOffsets[keyLength];
    long dataOffset = dataOffsets[keyLength];

    for (int probe = 0; probe < numSlots; probe++) {
      int slot = (int) ((hash + probe) % numSlots);
      indexBuffer.position(indexOffset + slot * slotSize);
      indexBuffer.get(slotBuffer, 0, slotSize);

      long offset = LongPacker.unpackLong(slotBuffer, keyLength);
      if (offset == 0) {
        return null;
      }
      if (isKey(slotBuffer, key)) {
        byte[] value = mMapData ? getMMapBytes(dataOffset + offset) : getDiskBytes(dataOffset + offset);
        return value;
      }
    }
    return null;
  }



//Read the data at the given offset, the data can be spread over multiple data buffers
  private byte[] getMMapBytes(long offset)
      throws IOException {
    //Read the first 4 bytes to get the size of the data
    ByteBuffer buf = getDataBuffer(offset);
    int maxLen = (int) Math.min(5, dataSize - offset);

    int size;
    if (buf.remaining() >= maxLen) {
      //Continuous read
      int pos = buf.position();
      size = LongPacker.unpackInt(buf);

      // Used in case of data is spread over multiple buffers
      offset += buf.position() - pos;
    } else {
      //The size of the data is spread over multiple buffers
      int len = maxLen;
      int off = 0;
      sizeBuffer.reset();
      while (len > 0) {
        buf = getDataBuffer(offset + off);
        int count = Math.min(len, buf.remaining());
        buf.get(sizeBuffer.getBuf(), off, count);
        off += count;
        len -= count;
      }
      size = LongPacker.unpackInt(sizeBuffer);
      offset += sizeBuffer.getPos();
      buf = getDataBuffer(offset);
    }

    //Create output bytes
    byte[] res = new byte[size];

    //Check if the data is one buffer
    if (buf.remaining() >= size) {
      //Continuous read
      buf.get(res, 0, size);
    } else {
      int len = size;
      int off = 0;
      while (len > 0) {
        buf = getDataBuffer(offset);
        int count = Math.min(len, buf.remaining());
        buf.get(res, off, count);
        offset += count;
        off += count;
        len -= count;
      }
    }

    return res;
  }


private ByteBuffer getDataBuffer(long index) {
    ByteBuffer buf = dataBuffers[(int) (index / segmentSize)];
    buf.position((int) (index % segmentSize));
    return buf;
  }


StorageReader的初始化过程

 StorageReader的初始化过程其实就是一个写入的逆向过程,怎么写入就怎么读取,整体的顺序如下:

  • 读取metaData的PalDB信息,包括key起始位移,value起始位移
  • 将key对应的内容映射到内存的mmap当中,不超过2GB。
  • 将value对应的内容映射到内存的mmap当中,根据segment大小进行切分
StorageReader(Configuration configuration, File file)
      throws IOException {
    path = file;
    config = configuration;
    
    //Config
    segmentSize = config.getLong(Configuration.MMAP_SEGMENT_SIZE);

    hashUtils = new HashUtils();

    // Check valid segmentSize
    if (segmentSize > Integer.MAX_VALUE) {
      throw new IllegalArgumentException(
          "The `" + Configuration.MMAP_SEGMENT_SIZE + "` setting can't be larger than 2GB");
    }

    //Open file and read metadata
    long createdAt = 0;
    FormatVersion formatVersion = null;
    FileInputStream inputStream = new FileInputStream(path);
    DataInputStream dataInputStream = new DataInputStream(new BufferedInputStream(inputStream));
    try {
      int ignoredBytes = -2;

      //Byte mark
      byte[] mark = FormatVersion.getPrefixBytes();
      int found = 0;
      while (found != mark.length) {
        byte b = dataInputStream.readByte();
        if (b == mark[found]) {
          found++;
        } else {
          ignoredBytes += found + 1;
          found = 0;
        }
      }

      //Version
      byte[] versionFound = Arrays.copyOf(mark, FormatVersion.getLatestVersion().getBytes().length);
      dataInputStream.readFully(versionFound, mark.length, versionFound.length - mark.length);

      formatVersion = FormatVersion.fromBytes(versionFound);
      if (formatVersion == null || !formatVersion.is(FormatVersion.getLatestVersion())) {
        throw new RuntimeException(
                "Version mismatch, expected was '" + FormatVersion.getLatestVersion() + "' and found '" + formatVersion
                        + "'");
      }

      //Time
      createdAt = dataInputStream.readLong();

      //Metadata counters
      keyCount = dataInputStream.readInt();
      keyLengthCount = dataInputStream.readInt();
      maxKeyLength = dataInputStream.readInt();

      //Read offset counts and keys
      indexOffsets = new int[maxKeyLength + 1];
      dataOffsets = new long[maxKeyLength + 1];
      keyCounts = new int[maxKeyLength + 1];
      slots = new int[maxKeyLength + 1];
      slotSizes = new int[maxKeyLength + 1];

      int maxSlotSize = 0;
      for (int i = 0; i < keyLengthCount; i++) {
        int keyLength = dataInputStream.readInt();

        keyCounts[keyLength] = dataInputStream.readInt();
        slots[keyLength] = dataInputStream.readInt();
        slotSizes[keyLength] = dataInputStream.readInt();
        indexOffsets[keyLength] = dataInputStream.readInt();
        dataOffsets[keyLength] = dataInputStream.readLong();

        maxSlotSize = Math.max(maxSlotSize, slotSizes[keyLength]);
      }

      slotBuffer = new byte[maxSlotSize];

      //Read serializers
      try {
        Serializers.deserialize(dataInputStream, config.getSerializers());
      } catch (Exception e) {
        throw new RuntimeException();
      }

      //Read index and data offset
      indexOffset = dataInputStream.readInt() + ignoredBytes;
      dataOffset = dataInputStream.readLong() + ignoredBytes;
    } finally {
      //Close metadata
      dataInputStream.close();
      inputStream.close();
    }

    //Create Mapped file in read-only mode
    mappedFile = new RandomAccessFile(path, "r");
    channel = mappedFile.getChannel();
    long fileSize = path.length();

    //Create index buffer
    indexBuffer = channel.map(FileChannel.MapMode.READ_ONLY, indexOffset, dataOffset - indexOffset);

    //Create data buffers
    dataSize = fileSize - dataOffset;

    //Check if data size fits in memory map limit
    if (!config.getBoolean(Configuration.MMAP_DATA_ENABLED)) {
      //Use classical disk read
      mMapData = false;
      dataBuffers = null;
    } else {
      //Use Mmap
      mMapData = true;

      //Build data buffers
      int bufArraySize = (int) (dataSize / segmentSize) + ((dataSize % segmentSize != 0) ? 1 : 0);
      dataBuffers = new MappedByteBuffer[bufArraySize];
      int bufIdx = 0;
      for (long offset = 0; offset < dataSize; offset += segmentSize) {
        long remainingFileSize = dataSize - offset;
        long thisSegmentSize = Math.min(segmentSize, remainingFileSize);
        dataBuffers[bufIdx++] = channel.map(FileChannel.MapMode.READ_ONLY, dataOffset + offset, thisSegmentSize);
      }
    }

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
阿里云服务器端口号设置
阿里云服务器初级使用者可能面临的问题之一. 使用tomcat或者其他服务器软件设置端口号后,比如 一些不是默认的, mysql的 3306, mssql的1433,有时候打不开网页, 原因是没有在ecs安全组去设置这个端口号. 解决: 点击ecs下网络和安全下的安全组 在弹出的安全组中,如果没有就新建安全组,然后点击配置规则 最后如上图点击添加...或快速创建.   have fun!  将编程看作是一门艺术,而不单单是个技术。
4502 0
LINQ 从数据库读数据生成 XML
本实例是一个从数据库里读出来省份,城市,地区的数据生成3个XML文件. 首先创建一个Linq to SQL的类, 然后将省份,城市,地区的表拖入类中: 然后新建一个类,加入代码如下:         public void ResetProvinces(string path)       ...
598 0
缓存键重复、冲突带来的数据读取错误
    第二次了,栽在这个问题上, 依然是出自缓存。     需要注意两个问题:     1.根据分类读的数据列表,缓存键需要带上分类ID/分类名称。     2.由于子模块比较多,缓存键命名的时候,整体搜一下是否已存在当前缓存键,否则容易出现多个模块公用一个缓存的现象,导致必有一处数据不对。
659 0
带你读《数据挖掘导论(原书第2版)》之一:绪论
本书所涵盖的主题包括:数据预处理、预测建模、关联分析、聚类分析、异常检测和避免错误发现。通过介绍每个主题的基本概念和算法,为读者提供将数据挖掘应用于实际问题所需的必要背景。
1430 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,阿里云优惠总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系.
5730 0
阿里云ECS云服务器初始化设置教程方法
阿里云ECS云服务器初始化是指将云服务器系统恢复到最初状态的过程,阿里云的服务器初始化是通过更换系统盘来实现的,是免费的,阿里云百科网分享服务器初始化教程: 服务器初始化教程方法 本文的服务器初始化是指将ECS云服务器系统恢复到最初状态,服务器中的数据也会被清空,所以初始化之前一定要先备份好。
10774 0
阿里云服务器远程登录用户名和密码的查询方法
阿里云服务器远程连接登录用户名和密码在哪查看?阿里云服务器默认密码是什么?云服务器系统不同默认用户名不同
444 0
使用JDBC操作数据库时,如何提升读取数据的性能?如何提升更新数据的性能?
要提升读取数据的性能,可以指定通过结果集(ResultSet)对象的setFetchSize()方法指定每次抓取的记录数(典型的空间换时间策略);要提升更新数据的性能可以使用PreparedStatement语句构建批处理,将若干SQL语句置于一个批处理中执行。
908 0
+关注
晴天哥
专注java技术,热爱长跑和阅读开源代码
343
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
文娱运维技术
立即下载
《SaaS模式云原生数据仓库应用场景实践》
立即下载
《看见新力量:二》电子书
立即下载