分布式分区

钝悟2022/6/14大约 33 分钟

分布式分区

什么是分区

分区通常是这样定义的，即每一条数据（或者每条记录，每行或每个文档）只属于某个特定分区。实际上，每个分区都可以视为一个完整的小型数据库，虽然数据库可能存在一些跨分区的操作。

在不同系统中，分区有着不同的称呼，例如它对应于 MongoDB, Elasticsearch 和 SolrCloud 中的 shard, HBase 的 region, Bigtable 中的 tablet, Cassandra 和 Riak 中的 vnode ，以及 Couch base 中的 vBucket。总体而言，分区是最普遍的术语。

特性

分布式分区具备以下核心特性：

特性	说明
水平扩展	通过增加节点线性扩展存储容量和吞吐量，突破单机瓶颈
数据局部性	每条数据只属于一个分区，查询可精确定位到目标分区，避免全表扫描
负载均衡	通过合理的分区策略，将数据和请求均匀分布到所有节点
透明性	对上层应用透明，分区的划分、迁移由系统自动管理
独立容错	每个分区可独立配置副本数，单个分区故障不影响其他分区
可再均衡	节点增删时，分区可动态迁移，保证集群负载持续均衡
与复制结合	分区通常与复制结合使用，每个分区在多个节点上存有副本

原理

分区的基本原理

分区的核心目标是将大量数据分散到多个节点上，每个节点只负责一部分数据。其基本原理如下：

graph TB
    Client[客户端请求] --> Router[路由层/分区感知]
    Router --> PartitionFunc[分区函数]
    PartitionFunc --> P1[分区 1<br/>节点 A]
    PartitionFunc --> P2[分区 2<br/>节点 B]
    PartitionFunc --> P3[分区 3<br/>节点 C]
    PartitionFunc --> P4[分区 4<br/>节点 D]

分区策略对比

策略	原理	优点	缺点	典型系统
基于关键字区间	按关键字排序，每个分区负责一段连续区间	支持高效的区间查询	易产生热点	`Bigtable`、`HBase`、`2.4 前 MongoDB`
基于哈希值	对关键字取哈希，按哈希值范围分区	数据分布均匀，避免热点	丧失区间查询能力	`Cassandra`、`MongoDB`（哈希分片）
一致性哈希	哈希值组成环，节点在环上占位	节点增删时迁移量小	可能数据分布不均	`Cassandra`、`Dynamo`
复合分区	主键第一部分哈希分区，其余部分排序	兼顾均匀分布和局部区间查询	复杂度较高	`Cassandra` 复合主键

分区再均衡流程

sequenceDiagram
    participant NodeA as 节点 A（原有分区1,2）
    participant NodeB as 节点 B（原有分区3,4）
    participant NodeC as 节点 C（新加入）
    participant Coordinator as 协调器

    Coordinator->>NodeC: 1. 新节点加入集群
    Coordinator->>NodeA: 2. 迁移部分分区（如分区2）
    Coordinator->>NodeB: 3. 迁移部分分区（如分区4）
    NodeA-->>NodeC: 4. 数据传输中（双写期）
    NodeB-->>NodeC: 5. 数据传输中（双写期）
    Coordinator->>Coordinator: 6. 更新路由表
    Coordinator->>NodeC: 7. 分区2,4 正式对外服务

为什么需要分区

数据量如果太大，单台机器进行存储和处理就会成为瓶颈，因此需要引入数据分区机制。

分区的目地是通过多台机器均匀分布数据和查询负载，避免出现热点。这需要选择合适的数据分区方案，在节点添加或删除时重新动态平衡分区。

数据分区与数据复制

分区通常与复制结合使用，即每个分区在多个节点都存有副本。这意味着某条记录属于特定的分区，而同样的内容会保存在不同的节点上以提高系统的容错性。

一个节点上可能存储了多个分区。每个分区都有自己的主副本，例如被分配给某节点，而从副本则分配在其他一些节点。一个节点可能既是某些分区的主副本，同时又是其他分区的从副本。

键－值数据的分区

分区的主要目标是将数据和查询负载均匀分布在所有节点上。如果节点平均分担负载，那么理论上 10 个节点应该能够处理 10 倍的数据量和 10 倍于单个节点的读写吞吐量（忽略复制）。

而如果分区不均匀，则会出现某些分区节点比其他分区承担更多的数据量或查询负载，称之为倾斜。倾斜会导致分区效率严重下降，在极端情况下，所有的负载可能会集中在一个分区节点上，这就意味着 10 个节点 9 个空闲，系统的瓶颈在最繁忙的那个节点上。这种负载严重不成比例的分区即成为系统热点。

避免热点最简单的方法是将记录随机分配给所有节点。这种方法可以比较均匀地分布数据，但是有一个很大的缺点：当视图读取特定的数据时，没有办法知道数据保存在哪个节点上，所以不得不并行查询所有节点。

可以改进上述方法。现在我们假设数据是简单的键-值数据模型，这意味着总是可以通过关键字来访问记录。

基于关键字区间分区

一种分区方式是为每个分区分配一段连续的关键字或者关键宇区间范围（以最小值和最大值来指示）。

关键字的区间段不一定非要均匀分布，这主要是因为数据本身可能就不均匀。

分区边界可以由管理员手动确定，或由数据库自动选择。采用这种分区策略的系统包括 Bigtable、HBase、RethinkDB、2.4 版本前的 MongoDB。

每个分区内可以按照关键字排序保存（参阅第 3 章的“ SSTables 和 LSM Trees ”）。这样可以轻松支持区间查询，即将关键字作为一个拼接起来的索引项从而一次查询得到多个相关记录。

然而，基于关键字的区间分区的缺点是某些访问模式会导致热点。如果关键字是时间戳，则分区对应于一个时间范围，所有的写入操作都集中在同一个分区（即当天的分区），这会导致该分区在写入时负载过高，而其他分区始终处于空闲状态。为了避免上述问题，需要使用时间戳以外的其他内容作为关键字的第一项。

基于关键字晗希值分区

对于上述数据倾斜和热点问题，许多分布式系统采用了基于关键字哈希函数的方式来分区。

一个好的哈希函数可以处理数据倾斜并使其均匀分布。用于数据分区目的的哈希函数不需要再加密方面很强：例如：Cassandra 和 MongoDB 使用 MD5，Voldemort 使用 Fowler-Noll-Vo。许多编程语言也有内置的简单哈希函数，但是要注意这些内置的哈希函数可能并不适合分区，例如，Java 的 Object.hashCode 和 Object#hash，同一个键在不同的进程中可能返回不同的哈希值。

一且找到合适的关键宇哈希函数，就可以为每个分区分配一个哈希范围（而不是直接作用于关键宇范围），关键字根据其哈希值的范围划分到不同的分区中。

这种方法可以很好地将关键字均匀地分配到多个分区中。分区边界可以是均匀间隔，也可以是伪随机选择（在这种情况下，该技术有时被称为一致性哈希）。

然而，通过关键宇哈希进行分区，我们丧失了良好的区间查询特性。即使关键字相邻，但经过哈希之后会分散在不同的分区中，区间查询就失去了原有的有序相邻的特性。在 MongoDB 中，如果启用了基于哈希的分片模式，则区间查询会发送到所有的分区上，而 Riak、Couchbase 和 Voldemort 干脆就不支持关键字上的区间查询。

Cassandra 则在两种分区策略之间做了一个折中。Cassandra 中的表可以声明为由多个列组成的复合主键。复合主键只有第一部分可用于哈希分区，而其他列则用作组合索引来对 Cassandra SSTable 中的数据进行排序。因此，它不支持在第一列上进行区间查询，但如果为第一列指定好了固定值，可以对其他列执行高效的区间查询。

组合索引为一对多的关系提供了一个优雅的数据模型。

负载倾斜与热点

基于哈希的分区方法可以减轻热点，但无住做到完全避免。一个极端情况是，所有的读／写操作都是针对同一个关键字，则最终所有请求都将被路由到同一个分区。

一个简单的技术就是在关键字的开头或结尾处添加一个随机数。只需一个两位数的十进制随机数就可以将关键字的写操作分布到 100 个不同的关键字上，从而分配到不同的分区上。但是，随之而来的问题是，之后的任何读取都需要些额外的工作，必须从所有 100 个关键字中读取数据然后进行合井。因此通常只对少量的热点关键字附加随机数才有意义；而对于写入吞吐量低的绝大多数关键字，这些都意味着不必要的开销。此外，还需要额外的元数据来标记哪些关键字进行了特殊处理。

分区与二级索引

二级索引通常不能唯一标识一条记录，而是用来加速特定值的查询。

二级索引是关系数据库的必要特性，在文档数据库中应用也非常普遍。但考虑到其复杂性，许多键-值存储（如 HBase 和 Voldemort）并不支持二级索引；但其他一些如 Riak 则开始增加对二级索引的支持。此外，二级索引技术也是 Solr 和 Elasticsearch 等全文索引服务器存在之根本。

二级索引带来的主要挑战是它们不能规整的地映射到分区中。有两种主要的方法来支持对二级索引进行分区：基于文档的分区和基于词条的分区。

基于文档分区的二级索引

在这种索引方法中，每个分区完全独立，各自维护自己的二级索引，且只负责自己分区内的文档而不关心其他分区中数据。每当需要写数据库时，包括添加，删除或更新文档等，只需要处理包含目标文档 ID 的那一个分区。因此文档分区索引也被称为本地索引，而不是全局索引。

这种查询分区数据库的方法有时也称为分散/聚集，显然这种二级索引的查询代价高昂。即使采用了并行查询，也容易导致读延迟显著放大。尽管如此，它还是广泛用于实践： MongoDB 、Riak、Cassandra、Elasticsearch 、SolrCloud 和 VoltDB 都支持基于文档分区二级索引。大多数数据库供应商都建议用户自己来构建合适的分区方案，尽量由单个分区满足二级索引查询，但现实往往难以如愿，尤其是当查询中可能引用多个二级索引时。

基于词条的二级索引分区

另一种方法，可以对所有的数据构建全局索引，而不是每个分区维护自己的本地索引。而且，为避免成为瓶颈，不能将全局索引存储在一个节点上，否则就破坏了设计分区均衡的目标。所以，全局索引也必须进行分区，且可以与数据关键字采用不同的分区策略。

词条分区以待查找的关键字本身作为索引。名字词条源于全文索引，term 指的是文档中出现的所有单词的集合。

可以直接通过关键词来全局划分索引，或者对其取哈希值。直接分区的好处是可以支持高效的区间查询；而采用哈希的方式则可以更均句的划分分区。

这种全局的词条分区相比于文档分区索引的主要优点是，它的读取更为高效，即它不需要采用 scatter/gather 对所有的分区都执行一遍查询，客户端只需要想包含词条的那一个分区发出读请求。然而全局索引的不利之处在于，写入速度较慢且非常复杂，主要因为单个文档的更新时，里面可能会涉及多个二级索引，而二级索引的分区又可能完全不同甚至在不同的节点上，由此势必引人显著的写放大。

理想情况下，索引应该时刻保持最新，即写入的数据要立即反映在最新的索引上。但是，对于词条分区来讲，这需要一个跨多个相关分区的分布式事务支持，写入速度会受到极大的影响，所以现有的数据库都不支持同步更新二级索引。

分区再均衡

集群节点数变化，数据规模增长等情况，都会导致分区的分布不均。要保持分区的均衡，势必要将数据和请求进行迁移，这样一个迁移负载的过程称为分区再均衡。

无论对于哪种分区方案，分区再平衡通常至少要满足：

平衡之后，负载、数据存储、读写请求等应该在集群范围更均匀地分布。
再平衡执行过程中，数据库应该可以继续正常提供读写服务。
避免不必要的负载迁移，以加快动态再平衡，井尽量减少网络和磁盘 I/O 影响。

动态再平衡的策略

为什么不用取模？

最好将哈希值划分为不同的区间范围，然后将每个区间分配给一个分区。

为什么不直接使用 mod？对节点数取模方法的问题是，如果节点数 N 发生了变化，会导致很多关键字需要从现有的节点迁移到另一个节点。

固定数量的分区

创建远超实际节点数的分区数，然后为每个节点分配多个分区。

接下来，如果集群中添加了一个新节点，该新节点可以从每个现有的节点上匀走几个分区，直到分区再次达到全局平衡。

选中的整个分区会在节点之间迁移，但分区的总数量仍维持不变，也不会改变关键字到分区的映射关系。这里唯一要调整的是分区与节点的对应关系。考虑到节点间通过网络传输数据总是需要些时间，这样调整可以逐步完成，在此期间，旧分区仍然可以接收读写请求。

原则上，也可以将集群中的不同的硬件配置因素考虑进来，即性能更强大的节点将分配更多的分区，从而分担更多的负载。

目前，Riak、Elasticsearch、Couchbase 和 Voldemort 都支持这种动态平衡方法。

使用该策略时，分区的数量往往在数据库创建时就确定好，之后不会改变。原则上也可以拆分和合并分区（稍后介绍），但固定数量的分区使得相关操作非常简单，因此许多采用固定分区策略的数据库决定不支持分区拆分功能。所以，在初始化时，已经充分考虑将来扩容增长的需求（未来可能拥有的最大节点数），设置一个足够大的分区数。而每个分区也有些额外的管理开销，选择过高的数字可能会有副作用。

动态分区

对于采用关键宇区间分区的数据库，如果边界设置有问题，最终可能会出现所有数据都挤在一个分区而其他分区基本为空，那么设定固定边界、固定数量的分区将非常不便：而手动去重新配置分区边界又非常繁琐。

因此，一些数据库如 HBase 和 RethinkDB 等采用了动态创建分区。当分区的数据增长超过一个可配的参数阔值（HBase 上默认值是 10GB），它就拆分为两个分区，每个承担一半的数据量。相反，如果大量数据被删除，并且分区缩小到某个阈值以下，则将其与相邻分区进行合井。该过程类似于 B 树的分裂操作。

每个分区总是分配给一个节点，而每个节点可以承载多个分区，这点与固定数量的分区一样。当一个大的分区发生分裂之后，可以将其中的一半转移到其他某节点以平衡负载。对于 HBase，分区文件的传输需要借助 HDFS。

动态分区的一个优点是分区数量可以自动适配数据总量。如果只有少量的数据，少量的分区就足够了，这样系统开销很小；如果有大量的数据，每个分区的大小则被限制在一个可配的最大值。

但是，需要注意的是，对于一个空的数据库，因为没有任何先验知识可以帮助确定分区的边界，所以会从一个分区开始。可能数据集很小，但直到达到第一个分裂点之前，所有的写入操作都必须由单个节点来处理，而其他节点则处于空闲状态。为了缓解这个问题，HBase 和 MongoDB 允许在一个空的数据库上配置一组初始分区（这被称为预分裂）。对于关键字区间分区，预分裂要求已经知道一些关键字的分布情况。

动态分区不仅适用于关键字区间分区，也适用于基于哈希的分区策略。MongoDB 从版本 2.4 开始，同时支持二者，井且都可以动态分裂分区。

按节点比例分区

采用动态分区策略，拆分和合并操作使每个分区的大小维持在设定的最小值和最大值之间，因此分区的数量与数据集的大小成正比关系。另一方面，对于固定数量的分区方式，其每个分区的大小也与数据集的大小成正比。两种情况，分区的数量都与节点数无关。

Cassandra 和 Ketama 则采用了第三种方式，使分区数与集群节点数成正比关系。换句话说，每个节点具有固定数量的分区。此时，当节点数不变时，每个分区的大小与数据集大小保持正比的增长关系；当节点数增加时，分区则会调整变得更小。较大的数据量通常需要大量的节点来存储，因此这种方法也使每个分区大小保持稳定。

当一个新节点加入集群时，它随机选择固定数量的现有分区进行分裂，然后拿走这些分区的一半数据量，将另一半数据留在原节点。随机选择可能会带来不太公平的分区分裂，但是当平均分区数量较大时（Cassandra 默认情况下，每个节点有 256 个分区），新节点最终会从现有节点中拿走相当数量的负载。Cassandra 在 3.0 时推出了改进算洁，可以避免上述不公平的分裂。

随机选择分区边界的前提要求采用基于哈希分区（可以从哈希函数产生的数字范围里设置边界）。这种方法也最符合本章开头所定义一致性哈希。一些新设计的哈希函数也可以以较低的元数据开销达到类似的效果。

自动与手动再平衡操作

动态平衡另一个重要问题我们还没有考虑：它是自动执行还是手动方式执行？

全自动式再平衡会更加方便，它在正常维护之外所增加的操作很少。但是，也有可能出现结果难以预测的情况。再平衡总体讲是个比较昂贵的操作，它需要重新路由请求井将大量数据从一个节点迁移到另一个节点。万一执行过程中间出现异常，会使网络或节点的负载过重，井影响其他请求的性能。

将自动平衡与自动故障检测相结合也可能存在一些风险。例如，假设某个节点负载过重，对请求的响应暂时受到影响，而其他节点可能会得到结论：该节点已经失效；接下来激活自动平衡来转移其负载。客观上这会加重该节点、其他节点以及网络的负荷，可能会使总体情况变得更槽，甚至导致级联式的失效扩散。

请求路由

当数据集分布到多个节点上，需要解决一个问题：当客户端发起请求时，如何知道应该连接哪个节点？如果发生了分区再平衡，分区与节点的对应关系随之还会变化。为了回答该问题，我们需要一段处理逻辑来感知这些变化，并负责处理客户端的连接。

这其实属于一类典型的服务发现问题，服务发现并不限于数据库，任何通过网络访问的系统都有这样的需求，尤其是当服务目标支持高可用时（在多台机器上有冗余配置）。

服务发现有以下处理策略：

允许客户端链接任意的节点（例如，采用循环式的负载均衡器）。如果某节点恰好拥有所请求的分区，则直接处理该请求：否则，将请求转发到下一个合适的节点，接收答复，并将答复返回给客户端。
将所有客户端的请求都发送到一个路由层，由后者负责将请求转发到对应的分区节点上。路由层本身不处理任何请求，它仅充一个分区感知的负载均衡器。
客户端感知分区和节点分配关系。此时，客户端可以直接连接到目标节点，而不需要任何中介。

许多分布式数据系统依靠独立的协调服务（如 ZooKeeper ）跟踪集群范围内的元数据。每个节点都向 ZooKeeper 中注册自己， ZooKeeper 维护了分区到节点的最终映射关系。其他参与者（如路由层或分区感知的客户端）可以向 ZooKeeper 订阅此信息。一旦分区发生了改变，或者添加、删除节点， ZooKeeper 就会主动通知路由层，这样使路由信息保持最新状态。

例如，HBase、SolrCloud 和 Kafka 也使用 ZooKeeper 来跟踪分区分配情况。MongoDB 有类似的设计，但它依赖于自己的配置服务器和 mongos 守护进程来充当路由层。

Cassandra 和 Riak 则采用了不同的方法，它们在节点之间使用 gossip 协议来同步群集状态的变化。请求可以发送到任何节点，由该节点负责将其转发到目标分区节点。这种方式增加了数据库节点的复杂性，但是避免了对 ZooKeeper 之类的外部协调服务的依赖。

并行查询执行

到目前为止，我们只关注了读取或写入单个关键字这样简单的查询（对于文档分区的二级索引，里面要求分散／聚集查询）。这基本上也是大多数 NoSQL 分布式数据存储所支持的访问类型。

然而对于大规模并行处理（massively parallel processing, MPP）这一类主要用于数据分析的关系数据库，在查询类型方面要复杂得多。典型的数据仓库查询包含多个联合、过滤、分组和聚合操作。MPP 查询优化器会将复杂的查询分解成许多执行阶段和分区，以便在集群的不同节点上井行执行。尤其是涉及全表扫描这样的查询操作，可以通过并行执行获益颇多。

小结

数据量如果太大，单台机器进行存储和处理就会成为瓶颈，因此需要引入数据分区机制。分区的目地是通过多台机器均匀分布数据和查询负载，避免出现热点。这需要选择合适的数据分区方案，在节点添加或删除时重新动态平衡分区。

两种主要的分区方法：

基于关键字区间的分区。先对关键字进行排序，每个分区只负责一段包含最小到最大关键字范围的一段关键字。对关键字排序的优点是可以支持高效的区间查询，但是如果应用程序经常访问与排序一致的某段关键字，就会存在热点的风险。采用这种方怯，当分区太大时，通常将其分裂为两个子区间，从而动态地再平衡分区。
哈希分区。将哈希函数作用于每个关键字，每个分区负责一定范围的哈希值。这种方法打破了原关键字的顺序关系，它的区间查询效率比较低，但可以更均匀地分配负载。采用哈希分区时，通常事先创建好足够多（但固定数量）的分区，让每个节点承担多个分区，当添加或删除节点时将某些分区从一个节点迁移到另一个节点，也可以支持动态分区。

混合上述两种基本方住也是可行的，例如使用复合键：键的一部分来标识分区，而另一部分来记录排序后的顺序。

二级索引也需要进行分区，有两种方法：

基于文档来分区二级索引（本地索引）。二级索引存储在与关键字相同的分区中，这意味着写入时我们只需要更新一个分区，但缺点是读取二级索引时需要在所有分区上执行 scatter/gather。
基于词条来分区二级索引（全局索引）。它是基于索引的值而进行的独立分区。二级索引中的条目可能包含来自关键字的多个分区里的记录。在写入时，不得不更新二级索引的多个分区；但读取时，则可以从单个分区直接快速提取数据。

最后，讨论了如何将查询请求路由到正确的分区，包括简单的分区感知负载均衡器，以及复杂的并行查询执行引擎。

应用场景

分布式分区在以下场景中应用广泛：

海量数据存储 - 当数据量超过单机存储能力时，通过分区将数据分散到多台机器。如 HBase 存储数百 TB 日志数据。
高并发读写 - 通过分区将读写负载分散到多个节点，提升整体吞吐量。如电商订单库按用户 ID 分片。
时序数据 - 监控指标、日志等时序数据按时间分区，便于按时间范围查询和过期数据清理。如 InfluxDB、TimescaleDB。
多租户系统 - SaaS 应用按租户 ID 分区，实现租户间数据物理隔离。
地理分布式部署 - 按地理位置分区，用户数据就近存储，降低访问延迟。如 Cassandra 多数据中心部署。
冷热数据分离 - 热数据存储在 SSD 节点，冷数据迁移到 HDD 节点，通过分区策略实现。

最佳实践

案例一：MySQL 分库分表（基于 ShardingSphere）

当单表数据量超过千万级，查询性能急剧下降，需要进行分库分表。Apache ShardingSphere 是一款优秀的分库分表中间件。

（1）引入依赖

<dependency>
    <groupId>org.apache.shardingsphere</groupId>
    <artifactId>shardingsphere-jdbc-core-spring-boot-starter</artifactId>
    <version>5.4.1</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.33</version>
</dependency>
<dependency>
    <groupId>com.zaxxer</groupId>
    <artifactId>HikariCP</artifactId>
    <version>4.0.3</version>
</dependency>

（2）配置分库分表规则

spring:
  shardingsphere:
    mode:
      type: Standalone
      repository:
        type: JDBC
    datasource:
      names: ds0,ds1
      ds0:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://127.0.0.1:3306/order_db_0?useSSL=false&characterEncoding=utf8
        username: root
        password: root
      ds1:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://127.0.0.1:3306/order_db_1?useSSL=false&characterEncoding=utf8
        username: root
        password: root
    rules:
      sharding:
        tables:
          t_order:
            # 真实数据节点：2 库 x 4 表 = 8 个分片
            actual-data-nodes: ds${0..1}.t_order_${0..3}
            # 分库策略：按 user_id 取模
            database-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: db-mod
            # 分表策略：按 order_id 取模
            table-strategy:
              standard:
                sharding-column: order_id
                sharding-algorithm-name: table-mod
            # 主键生成策略
            key-generate-strategy:
              column: order_id
              key-generator-name: snowflake
        sharding-algorithms:
          db-mod:
            type: MOD
            props:
              sharding-count: 2
          table-mod:
            type: MOD
            props:
              sharding-count: 4
        key-generators:
          snowflake:
            type: SNOWFLAKE
            props:
              worker-id: 1
    props:
      sql-show: true  # 打印分片 SQL 日志，便于调试

（3）实体类与 Mapper

import lombok.Data;

@Data
public class Order {
    private Long orderId;
    private Long userId;
    private String productCode;
    private Integer quantity;
    private java.math.BigDecimal amount;
    private java.time.LocalDateTime createTime;
}

import org.apache.ibatis.annotations.*;

@Mapper
public interface OrderMapper {

    @Insert("INSERT INTO t_order (order_id, user_id, product_code, quantity, amount, create_time) " +
            "VALUES (#{orderId}, #{userId}, #{productCode}, #{quantity}, #{amount}, #{createTime})")
    int insert(Order order);

    @Select("SELECT * FROM t_order WHERE order_id = #{orderId} AND user_id = #{userId}")
    Order selectByOrderId(@Param("orderId") Long orderId, @Param("userId") Long userId);
}

说明：ShardingSphere 对业务代码完全透明，开发者像操作单表一样编写 SQL，中间件自动完成路由、改写和结果归并。分库键选择 user_id 保证同一用户的订单在同一库中，分表键选择 order_id 保证订单均匀分布。

案例二：Kafka 分区设计与消费者分配

Kafka 的分区（Partition）是其并行处理和水平扩展的基础，合理的分区数和分配策略对性能至关重要。

（1）创建多分区 Topic

# 创建 6 个分区、3 个副本的 topic
kafka-topics.sh --create \
  --bootstrap-server 127.0.0.1:9092 \
  --topic order-events \
  --partitions 6 \
  --replication-factor 3

（2）Java 生产者指定分区策略

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class KafkaPartitionProducer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                  "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                  "org.apache.kafka.common.serialization.StringSerializer");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            // 方式一：指定 key，Kafka 会对 key 取 hash 自动分区
            // 同一 key 的消息总是进入同一分区，保证顺序性
            ProducerRecord<String, String> record1 =
                new ProducerRecord<>("order-events", "user-1001", "订单创建事件");
            producer.send(record1, (metadata, e) -> {
                if (e == null) {
                    System.out.printf("消息发送成功: partition=%d, offset=%d%n",
                        metadata.partition(), metadata.offset());
                } else {
                    e.printStackTrace();
                }
            });

            // 方式二：直接指定分区号
            ProducerRecord<String, String> record2 =
                new ProducerRecord<>("order-events", 3, "user-1002", "订单支付事件");
            producer.send(record2);

            // 方式三：自定义分区器
            // props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomPartitioner.class.getName());
        }
    }
}

（3）自定义分区器

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

import java.util.List;
import java.util.Map;

public class CustomPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        // 按 key 的 hash 取模分区
        if (keyBytes == null) {
            // 无 key 时轮询
            return Math.abs(value.hashCode()) % numPartitions;
        }
        // 将 VIP 用户的消息路由到前 2 个分区（高优先级）
        if (key.toString().startsWith("vip-")) {
            return Math.abs(key.hashCode()) % 2;
        }
        return Math.abs(key.hashCode()) % numPartitions;
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

说明：Kafka 分区数建议设置为消费者数量的整数倍，保证负载均匀。分区数确定后不能减少（只能增加），需提前规划。

案例三：Redis Cluster 分片部署

Redis Cluster 通过哈希槽（Hash Slot）实现分区，共 16384 个槽位，分布到多个节点。

（1）集群配置（6 节点：3 主 3 从）

# 节点配置文件 redis-7000.conf
port 7000
cluster-enabled yes
cluster-config-file nodes-7000.conf
cluster-node-timeout 5000
appendonly yes

（2）启动节点并创建集群

# 启动 6 个节点（7000-7005）
for port in 7000 7001 7002 7003 7004 7005; do
    redis-server redis-${port}.conf
done

# 创建集群，3 主 3 从，每个主节点 1 个从节点
redis-cli --cluster create \
  127.0.0.1:7000 127.0.0.1:7001 127.0.0.1:7002 \
  127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 \
  --cluster-replicas 1

（3）Java 客户端操作

import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;

public class RedisClusterExample {

    public static void main(String[] args) {
        RedisClusterClient client = RedisClusterClient.create(
            "redis://127.0.0.1:7000"
        );

        try (StatefulRedisClusterConnection<String, String> connection = client.connect()) {
            RedisAdvancedClusterCommands<String, String> sync = connection.sync();

            // Redis 自动根据 key 的 hash slot 路由到对应节点
            sync.set("user:1001", "张三");
            sync.set("user:1002", "李四");
            sync.set("order:2001", "订单数据");

            System.out.println("user:1001 = " + sync.get("user:1001"));
            System.out.println("user:1002 = " + sync.get("user:1002"));

            // 使用 hash tag 强制多个 key 到同一槽位（支持事务和 MGET）
            sync.set("{user:1001}:profile", "个人信息");
            sync.set("{user:1001}:settings", "用户设置");

            // 以下操作可以跨槽位执行（因 hash tag 相同）
            System.out.println("profile = " + sync.get("{user:1001}:profile"));
            System.out.println("settings = " + sync.get("{user:1001}:settings"));
        } finally {
            client.shutdown();
        }
    }
}

说明：Redis Cluster 使用 CRC16 算法计算 key 的哈希槽，槽位范围 0-16383。使用 {} 括起来的部分作为哈希计算依据，称为 Hash Tag，可将相关 key 路由到同一节点，便于执行事务和 MGET 等批量操作。

常见问题

问题一：分片后热点问题（数据倾斜）

问题描述

MySQL 分库分表后，发现某个分片数据量和请求量远高于其他分片，导致该节点 CPU 和 I/O 持续高位，而其他节点资源闲置。

原因分析

分片键选择不当：如按时间字段分片，近期数据集中在同一分片。
数据本身倾斜：某些热门用户（如大 V）产生大量数据，集中在一个分片。
哈希函数分布不均：哈希函数设计不合理，导致数据分布不均匀。

解决方案

选择基数大、分布均匀的字段作为分片键，如 user_id 而非 create_time。
对热点 key 添加随机前缀，打散到多个分片：

import java.util.concurrent.ThreadLocalRandom;

public class HotKeySharder {

    /**
     * 对热点 key 添加随机后缀，打散到 10 个分片
     */
    public static String shardHotKey(String originalKey) {
        int shard = ThreadLocalRandom.current().nextInt(10);
        return originalKey + ":" + shard;
    }

    /**
     * 查询时需聚合所有分片的数据
     */
    public static String[] getAllShardKeys(String originalKey) {
        String[] keys = new String[10];
        for (int i = 0; i < 10; i++) {
            keys[i] = originalKey + ":" + i;
        }
        return keys;
    }
}

使用一致性哈希，减少节点增删时的数据迁移：

# Cassandra 配置一致性哈希分区
# cassandra.yaml
partitioner: org.apache.cassandra.dht.Murmur3Partitioner
num_tokens: 16  # 虚拟节点数，值越大分布越均匀

问题二：分区再均衡期间服务不可用

问题描述

Elasticsearch 集群新增节点后触发 shard rebalancing，期间集群状态变为 yellow 或 red，查询延迟剧增甚至超时。

原因分析

再均衡并发度过高：默认配置下同时迁移多个分片，占用大量网络和磁盘 I/O。
节点资源不足：新节点未预热，磁盘 I/O 瓶颈导致恢复速度慢。
再均衡期间触发熔断器：大量请求堆积导致熔断。

解决方案

限制再均衡并发度和速率：

# elasticsearch.yml
cluster:
  routing:
    allocation:
      # 同时允许的分片恢复并发数
      node_concurrent_recoveries: 2
      # 每秒恢复的数据量上限（字节）
      node_initial_primaries_recoveries: 4
      cluster_concurrent_rebalance: 2
      # 再均衡速率（MB/s）
      disk.watermark.low: 85%
      disk.watermark.high: 90%

在低峰期执行再均衡，并手动控制分片迁移：

# 禁用自动分片分配
curl -X PUT "localhost:9200/_cluster/settings" -H 'Content-Type: application/json' -d'
{
  "transient": {
    "cluster.routing.rebalance.enable": "none"
  }
}'

# 手动将分片迁移到新节点
curl -X POST "localhost:9200/_cluster/reroute" -H 'Content-Type: application/json' -d'
{
  "commands": [
    {
      "move": {
        "index": "order-index",
        "shard": 0,
        "from_node": "node-1",
        "to_node": "node-3"
      }
    }
  ]
}'

# 完成后恢复自动分配
curl -X PUT "localhost:9200/_cluster/settings" -H 'Content-Type: application/json' -d'
{
  "transient": {
    "cluster.routing.rebalance.enable": "all"
  }
}'

预热新节点：先将只读副本分配到新节点，待数据完整后再开放写入。

问题三：跨分区查询性能差

问题描述

分库分表后，某些查询（如按非分片键查询、统计聚合）需要扫描所有分片，查询耗时是单表的数倍。

原因分析

分片键决定了数据的分布，非分片键查询无法定位到特定分片，只能广播到所有分片执行（scatter-gather），然后合并结果。分区数越多，查询放大越严重。

解决方案

引入二级索引或映射表，建立非分片键到分片键的映射：

-- 创建映射表（不分片，存储在单独的库中）
CREATE TABLE order_user_mapping (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    order_id BIGINT NOT NULL COMMENT '订单ID',
    user_id BIGINT NOT NULL COMMENT '用户ID（分片键）',
    UNIQUE KEY uk_order_id (order_id),
    KEY idx_user_id (user_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 查询时：先查映射表获取 user_id，再精确路由到对应分片
SELECT user_id FROM order_user_mapping WHERE order_id = 10086;
-- 假设查到 user_id = 1001
SELECT * FROM t_order WHERE order_id = 10086 AND user_id = 1001;

使用 Elasticsearch 等搜索引擎构建全局索引，支持复杂查询：

// 写入 MySQL 的同时异步同步到 ES
@Async
public void syncToElasticsearch(Order order) {
    OrderDoc doc = new OrderDoc();
    doc.setOrderId(order.getOrderId());
    doc.setUserId(order.getUserId());
    doc.setProductCode(order.getProductCode());
    doc.setAmount(order.getAmount());
    doc.setCreateTime(order.getCreateTime());
    // ES 支持任意字段查询，不受分片键限制
    orderRepository.save(doc);
}

// 复杂查询走 ES
public List<OrderDoc> searchOrders(String productCode, BigDecimal minAmount) {
    BoolQueryBuilder query = QueryBuilders.boolQuery()
        .must(QueryBuilders.matchQuery("productCode", productCode))
        .must(QueryBuilders.rangeQuery("amount").gte(minAmount));
    return orderRepository.search(query);
}

合理设计 CQL（Cassandra Query Language），利用复合主键支持多种查询模式：

-- Cassandra 复合主键设计
CREATE TABLE orders (
    user_id bigint,           -- 分区键（哈希分区）
    order_id timeuuid,        -- 聚簇列（区间查询）
    product_code text,
    amount decimal,
    create_time timestamp,
    PRIMARY KEY ((user_id), order_id)
) WITH CLUSTERING ORDER BY (order_id DESC);

-- 查询1：按用户查最近订单（精确路由分区）
SELECT * FROM orders WHERE user_id = 1001 LIMIT 10;

-- 查询2：按用户和时间范围查（精确路由分区 + 区间查询）
SELECT * FROM orders
WHERE user_id = 1001
  AND order_id > minTimeuuid('2024-01-01 00:00:00');

参考资料

《数据密集型应用系统设计》 - 这可能是目前最好的分布式存储书籍，强力推荐【进阶】