数据库中间件 MyCAT 源码解析 —— 分片结果合并(一)

本人花费半年的时间总结的《Java面试指南》已拿腾讯等大厂offer,已开源在github ,欢迎star!

转载声明:转载请注明出处,本技术博客是本人原创文章

本文GitHub https://github.com/OUYANGSIHAI/JavaInterview 已收录,这是我花了6个月总结的一线大厂Java面试总结,本人已拿大厂offer,欢迎star

原文链接:blog.ouyangsihai.cn >> 数据库中间件 MyCAT 源码解析 —— 分片结果合并(一)

1. 概述

相信很多同学看过 MySQL 各种优化的文章,里面 99% 会提到:单表数据量大了,需要进行分片(水平拆分 or 垂直拆分)。分片之后,业务上必然面临的场景:跨分片的数据合并。今天我们就一起来瞅瞅 MyCAT 是如何实现分片结果合并。

跨分片查询大体流程如下:

和 《【单库单表】查询》 不同的两个过程:

  • 【2】多分片执行 SQL
  • 【4】合并多分片结果

下面,我们来逐条讲解这两个过程。

2. 多分片执行 SQL

经过 SQL 解析后,计算出需要执行 SQL 的分片节点,遍历分片节点发送 SQL 进行执行。

核心代码:

  • MultiNodeQueryHandler.java#execute(…)

SQL 解析 详细过程,我们另开文章,避免内容过多,影响大家对 分片结果合并 流程和逻辑的理解。

3. 合并多分片结果

 

handle_response

和 《【单库单表】查询》 不同,多个分片节点都会分别响应 记录头(header) 和 记录行(row) 。在开始分析 MyCAT 是怎么合并多分片结果之前,我们先来回想下 SQL 的执行顺序。

12345678
FROM       // [1] 选择表 WHERE      // [2] 过滤表 GROUP BY   // [3] 分组SELECT     // [4] 普通字段,max / min / avg / sum / count 等函数,distinctHAVING     // [5] 再过滤表ORDER BY   // [6] 排序LIMIT      // [7] 分页

FROM      
// [1] 选择表 WHERE      
// [2] 过滤表 GROUP BY  
// [3] 分组SELECT    
// [4] 普通字段,max / min / avg / sum / count 等函数,distinctHAVING    
// [5] 再过滤表ORDER BY  
// [6] 排序LIMIT      
// [7] 分页

3.1 记录头(header)

多个分片节点响应时,会响应多次 记录头(header) 。MyCAT 在实际处理时,只处理第一个返回的 记录头(header) 。因此,在使用时要保证表的 Schema 相同。

分片节点响应的 记录头(header) 可以直接返回 MySQL Client 吗?答案是不可以。 AVG函数 是特殊情况,MyCAT 需要将  AVG 拆成  SUM +  COUNT 进行计算。举个例子:

123
// [1] MySQL Client = MyCAT : SELECT AVG(age) FROM student;  // [2] MyCAT = MySQL Server : SELECT SUM(age) AS AVG0SUM, COUNT(age) AS AVG0COUNT FROM student;  // [3] 最终:AVG(age) = SUM(age) AS AVG0SUM / COUNT(age)

// [1] MySQL Client = MyCAT : SELECT AVG(age) FROM student;
// [2] MyCAT = MySQL Server : SELECT SUM(age) AS AVG0SUM, COUNT(age) AS AVG0COUNT FROM student;
// [3] 最终:AVG(age) = SUM(age) AS AVG0SUM / COUNT(age)

核心代码:

  • MultiNodeQueryHandler.java#fieldEofResponse(…)。

3.2 记录行(row)

3.1 AbstractDataNodeMerge

MyCAT 对分片结果合并通过  AbstractDataNodeMerge 子类来完成。

AbstractDataNodeMerge :

  • -packs :待合并记录行(row)队列。队列尾部插入  END_FLAG_PACK 表示队列已结束。
  • -running :合并逻辑是否正在执行中的标记。
  • ~onRowMetaData(…) :根据记录列信息(ColMeta)构建对应的排序组件和聚合组件。需要子类进行实现。
  • ~onNewRecord(…) :插入记录行(row) 到  packs
  • ~outputMergeResult(…) :插入  END_FLAG_PACK 到  packs
  • ~run(…) :执行合并分片结果逻辑,并将合并结果返回给 MySQL Client。需要子类进行实现。

通过  running 标记保证同一条 SQL 同时只有一个线程正在执行,并且不需要等到每个分片结果都返回就可以执行聚合逻辑。当然,排序逻辑需要等到所有分片结果都返回才可以执行。

核心代码:

  • AbstractDataNodeMerge.java
  • DataNodeMergeManager.java#run(…)

3.2 DataNodeMergeManager

AbstractDataNodeMerge 有两种子类实现:

  • DataMergeService :基于堆内内存合并分片结果。
  • DataNodeMergeManager :基于堆外内存合并分片结果。

目前官方默认配置使用  DataNodeMergeManager。主要有如下优点:

  1. 可以使用更大的内存空间。当并发量大或者数据量大时,更大的内存空间意味着更好的性能。
  2. 减少 GC 暂停时间。记录行(row)对象小且重用性很低,需要能够进行类似 C / C++ 的自主内存释放。
  3. 更快的内存复制和读取速度,对排序和聚合带来很好的提速。

如果对堆外内存不太了解,推荐阅读如下文章:

  1. 《从0到1起步-跟我进入堆外内存的奇妙世界》
  2. 《堆内内存还是堆外内存?》
  3. 《JAVA堆外内存》
  4. 《JVM源码分析之堆外内存完全解读》

本文主要分析  DataNodeMergeManager 实现, DataMergeService 可以自己阅读或者等待后续文章(😈欢迎订阅我的公众号噢)。

DataNodeMergeManager 有三个组件:

  • globalSorter : UnsafeExternalRowSorter = 实现记录行(row)合并并排序逻辑。
  • globalMergeResult : UnsafeExternalRowSorter = 实现记录行(row)合并不排序逻辑。
  • unsafeRowGrouper :  UnsafeRowGrouper = 实现记录行(row)聚合逻辑。

DataNodeMergeManager#run(...) 逻辑如下:

  • [1] 写入记录行(row)到  UnsafeRow
  • [2] 根据情况将  UnsafeRow 插入对应组件。
  • [3] 当所有  UnsafeRow 插入完后,根据情况使用组件聚合、排序。

核心代码:

  • DataNodeMergeManager.java。

🙃看到这里,可能很多同学都有点懵逼,问题不大,我们继续往下瞅。

3.3 UnsafeRow

记录行(row)写到  UnsafeRow 的  baseObject 属性,结构如下:

  • 拆分成三个区域,每个区域按照格子记录信息,每个格子 64bits(8 Bytes)。

  • 记录行(row)按照字段顺序位置记录到  baseObject

  • [1] 空标记位区域 :标记字段对应的值是否为 NULL。
    • 当字段对应的值为 NULL 时,其对应的字段顺序对应的 bit 设置为 1。举个例子,第 0 个位置字段为 NULL,则第一个格子对应的 64 bits 从右边第一个 bit 设置为 1。

    • 因为每个格子是 64 bits,每 64 个字段占用一个格子,不满一个格子,按照一个格子计算。因此,该区域的长度( bitSetWidthInBytes) = 字段占用的格子数 * 64 bits。

    • 每个字段记录 [2]区域的位置 =  baseOffset +  bitSetWidthInBytes + 8 Bytes * 字段顺序。

    • 占用一个格子,前 32 bits 为 [3]区域的位置,后 32 bits 为字段对应的值长度。

    • 每个字段对应的值占用格子数 = 字段对应的值长度 / 8 Byte,如果无法整除再 + 1。

    • 因为字段对应的值可能无法刚好占满每个格子,未使用的 bit 用 0 占位。

    写入  UnsafeRow,MyCAT 可以顺序访问每个字段,而不需要在记录行(row)进行遍历。

    🙃日常开发使用位操作的机会比较少,可能较为难理解,需要反复理解下,相信会获得很大启发。恩,该部分代码引用自开源运算框架  Spark,是不是更加有动力列😈。

    核心代码:

    • UnsafeRow.java
    • BufferHolder.java
    • UnsafeRowWriter.java

    3.4 UnsafeExternalRowSorter

    如果使用 Java 实现  SELECT * FROM student ORDER BY age desc, nickname asc,不考虑算法优化的情况下,我们可以简单如下实现:

    12345678
    Collections.sort(students,new ComparatorComparable(){      @Override     public int compare(Student o1,Student o2){                  int cmp=compare(o2.age,o1.age);                  return cmp!=0?cmp:compare(o1.nickname,o2.nickname);              }         }}        );

    Collections.sort(students,new ComparatorComparable(){
      @Override
         public int compare(Student o1,Student o2){
              int cmp=compare(o2.age,o1.age);
              return cmp!=0?cmp:compare(o1.nickname,o2.nickname);
          }
     }}
    );

    从功能上, UnsafeExternalRowSorter 是这么实现排序逻辑。当然肯定的是,不是这么“简单”的实现。

    UnsafeRow 会写入到两个地方:

    1. ListMemoryBlock :内存块数组。当前  MemoryBlock 无法容纳写入的  UnsafeRow时,生成新的  MemoryBlock 提供写入。每条  UnsafeRow 存储在  MemoryBlock 由 长度 + 字节内容 组成。
    2. `LongArray` :每条 `UnsafeRow` 存储在 `LongArray` 由两部分组成:address + prefix。
      • address : UnsafeRow 存储在  ListMemoryBlock 的位置。前 13 bits 记录所在  MemoryBlock 的 index,后 51 bit 记录在  MemoryBlock 的 offset。
      • prefix : UnsafeRow 第一个排序字段值前 64 bits 计算的值。

    UnsafeExternalRowSorter 排序实现方式 :提供 TimSort 和 RadixSort 两种排序算法,前者为默认实现。TimSort 折半查找时,使用  LongArray,先比较  prefix,若相等,则顺序对比每个排序字段直到不等,提升计算效率。插入操作在  LongArray 操作, ListMemoryBlock 只作为原始数据。

    另外,当需要排序特别大的数据量时,会使用存储数据到文件进行排序。限于笔者暂时未阅读该处源码,后续会另开文章分析。🙂

    核心源码:

    • UnsafeExternalRowSorter.java
    • UnsafeExternalRowSorter.java
    • TimSort.java

    3.5 UnsafeRowGrouper

    如果使用 Java 实现  SELECT nickname, COUNT(*) FROM student group by nickname,不考虑算法优化的情况下,我们可以简单如下实现:

    1234567891011121314
    MapString, ListObjectmap=new HashMap();// 聚合for (student : students) {    if (map.contains(student.nickname)) {            map.put(student.nickname, map.get(student.nickname).get(1) + 1);       } else {            ListObject value = new Array();           value.add(nickname);           value.add(1);           map.put(student.nickname, value);        } }// 输出for (value : map.values) {        System.out.println(value);}

    MapString, ListObjectmap=new HashMap();// 聚合
    for (student : students) {
       if (map.contains(student.nickname)) {
           map.put(student.nickname, map.get(student.nickname).get(1) + 1);
       } else {
           ListObject value = new Array();
           value.add(nickname);
           value.add(1);
           map.put(student.nickname, value);
       }
    }// 输出
    for (value : map.values) {
       System.out.println(value);
    }

    从功能上, UnsafeRowGrouper 是这么实现排序逻辑。当然肯定的是,也不是这么“简单”的实现。

    😈具体怎么实现的呢?我们在《MyCAT 源码解析 —— 分片结果合并(二)》继续分析。

    4. 救护中心

    看到此处的应该是真爱吧?!如果内容上有什么错误或者难懂的地方,可以关注我的微信公众号给我留言,我会很认真的逐条解答的。“万一”觉得本文还可以,希望转发到朋友圈让更多的人看到。

    最后的最后,感谢耐心阅读本文的同学!

     

     

本人花费半年的时间总结的《Java面试指南》已拿腾讯等大厂offer,已开源在github ,欢迎star!

转载声明:转载请注明出处,本技术博客是本人原创文章

本文GitHub https://github.com/OUYANGSIHAI/JavaInterview 已收录,这是我花了6个月总结的一线大厂Java面试总结,本人已拿大厂offer,欢迎star

原文链接:blog.ouyangsihai.cn >> 数据库中间件 MyCAT 源码解析 —— 分片结果合并(一)