标签：datax 查询 virt 分布式系统 pack generator parameter 方案 压力
数据分表小结标签： sharding 数据表拆分
最近一段时间内结束了数据库表拆分项目，这里做个简单的小结。
本次拆分主要包括订单和优惠券两大块，这两块都是覆盖全集团所有分子公司所有业务线。随着公司的业务飞速发展，不管是存储的要求，还是写入、读取的性都基本上到了警戒水位。
订单是交易的核心，优惠券是营销的核心，这两块基本上是整个平台的正向最核心部分。为了支持未来三到五年的快速发展，我们需要对数据进行拆分。
数据库表拆分业内已经有很多成熟方案，已经不是什么高深的技术，基本上是纯工程化的流程，但是能有机会进行实际的操刀一把机会还是难得，所以非常有必要做个总结。
由于分库分表包含的技术选型和方式方法多种多样，这篇文章不是罗列和汇总介绍各种方法，而是总结我们在实施分库分表过程中的一些经验。
根据业务场景判断，我们主要是做水平拆分，做逻辑 DB 拆分，考虑到未来数据库写入瓶颈可以将一组 sharding 表直接迁移进分库中。
分库、分表会带来很多的后遗症，会使整个系统架构变的复杂。分的好与不好最关键就是如何寻找那个 sharding key，如果这个 sharding key 刚好是业务维度上的分界线就会直接提升性能和改善复杂度，否则就会有各种脚手架来支撑，系统也就会变得复杂。
比如订单系统中的用户ID、订单type、商家ID、渠道ID，优惠券系统中的批次ID、渠道ID、机构ID 等，这些都是潜在的 sharding key。
如果刚好有这么一个 sharding key 存在后面处理路由（routing）就会很方便，否则就需要一些大而全的索引表来处理 OLAP 的查询。
一旦 sharding 之后首先要面对的问题就是查询时排序分页问题。
原来在一个数据库表中处理排序分页是比较方便的，sharding 之后就会存在多个数据源，这里我们将多个数据源统称为分片。
想要实现多分片排序分页就需要将各个片的数据都汇集起来进行排序，就需要用到 归并排序 算法。这些数据在各个分片中可以做到有序的（输出有序），但是整体上是无序的。
我们看个简单的例子：
shard node 1: {1、3、5、7、9}
shard node 2: {2、4、6、8、10}这是做 奇偶 sharding 的两个分片，我们假设分页参数设置为每页4条，当前第1页，参数如下：
pageParameter：pageSize：4、currentPage：1最乐观情况下我们需要分别读取两个分片节点中的前两条：
shard node 1: {1、3}
shard node 2: {2、4}排序完刚好是 {1、2、3、4}，但是这种场景基本上不太可能出现，假设如下分片节点数据：
shard node 1: {7、9、11、13、15}
shard node 2: {2、4、6、8、10、12、14}我们还是按照读取每个节点前两条肯定是错误的，因为最悲观情况下也是最真实的情况就是排序完后所有的数据都来自一个分片。所以我们需要读取每个节点的 pageSize 大小的数据出来才有可能保证数据的正确性。
这个例子只是假设我们的查询条件输出的数据刚好是均等的，真实的情况一定是各种各样的查询条件筛选出来的数据集合，此时这个数据一定不是这样的排列方式，最真实的就是最后者这种结构。
我们以此类推，如果我们的 currentPage:1000 那么会出现什么问题，我们需要每个 sharding node 读取 4000(1000*4=4000) 条数据出来排序，因为最悲观情况下有可能所有的数据均来自一个 sharding node 。
这样无限制的翻页下去，处理排序分页的机器肯定会内存撑爆，就算不撑爆一定会触发性能瓶颈。
这个简单的例子用来说明分片之后，排序分页带来的现实问题，这也有助于我们理解分布式系统在做多节点排序分页时为什么有最大分页限制。
一个庞大的数据集会通过多种方式进行数据拆分，按机构、按时间、按渠道等等，拆分在不同的数据源中。一般的深分页问题我们可以通过改变查询条件来平滑解决，但是这种方案并不能解决所有的业务场景。
比如，我们有一个订单列表，从C端用户来查询自己的订单列表数据量不会很大，但是运营后台系统可能面对全平台的所有订单数据量，所以数据量会很大。
改变查询条件有两种方式，一种是显示的设置，尽量缩小查询范围，这种设置一般都会优先考虑，比如时间范围、支付状态、配送状态等等，通过多个叠加条件就可以横竖过滤出很小一部分数据集。
那么第二种条件为隐式设置。比如订单列表通常是按照订单创建时间来排序，那么当翻页到限制的条件时，我们可以改变这个时间。
sharding node 1:
orderID createDateTime
100000 2018-01-10 10:10:10
200000 2018-01-10 10:10:11
300000 2018-01-10 10:10:12
400000 2018-01-10 10:10:13
500000 2018-01-20 10:10:10
600000 2018-01-20 10:10:11
700000 2018-01-20 10:10:12sharding node 2:
orderID createDateTime
110000 2018-01-11 10:10:10
220000 2018-01-11 10:10:11
320000 2018-01-11 10:10:12
420000 2018-01-11 10:10:13
520000 2018-01-21 10:10:10
620000 2018-01-21 10:10:11
720000 2018-01-21 10:10:12我们假设上面是一个订单列表，orderID 订单号大家就不要在意顺序性了。因为 sharding 之后所有的 orderID 都会由发号器统一发放，多个集群多个消费者同时获取，但是创建订单的速度是不一样的，所以顺序性已经不存在了。
上面的两个 sharding node 基本上订单号是交叉的，如果按照时间排序 node 1 和 node 2 是要交替获取数据。
比如我们的查询条件和分页参数：
where createDateTime>‘2018-01-11 00:00:00‘pageParameter：pageSize：5、currentPage：1获取的结果集为：
orderID createDateTime
100000 2018-01-10 10:10:10
200000 2018-01-10 10:10:11
300000 2018-01-10 10:10:12
400000 2018-01-10 10:10:13
110000 2018-01-11 10:10:10前面 4 条记录来自 node 1 后面 1 条数据来自 node 2 ，整个排序集合为：
sharding node 1:
orderID createDateTime
100000 2018-01-10 10:10:10
200000 2018-01-10 10:10:11
300000 2018-01-10 10:10:12
400000 2018-01-10 10:10:13
500000 2018-01-20 10:10:10
sharding node 2:
orderID createDateTime
110000 2018-01-11 10:10:10
220000 2018-01-11 10:10:11
320000 2018-01-11 10:10:12
420000 2018-01-11 10:10:13
520000 2018-01-21 10:10:10按照这样一直翻页下去每翻页一次就需要在 node 1 、node 2 多获取 5 条数据。这里我们可以通过修改查询条件来让整个翻页变为重新查询。
where createDateTime>‘2018-01-11 10:10:13‘因为我们可以确定在 ‘2018-01-11 10:10:13’ 时间之前所有的数据都已经查询过，但是为什么时间不是从 ‘2018-01-21 10:10:10’ 开始，因为我们要考虑并发情况，在 1s 内会有多个订单进来。
这种方式是实现最简单，不需要借助外部的计算来支撑。这种方式有一个问题就是要想重新计算分页的时候不丢失数据就需要保留原来一条数据，这样才能知道开始的时间在哪里，这样就会在下次的分页中看到这条时间。但是从真实的深分页场景来看也可以忽略，因为很少有人会一页一页一直到翻到500页，而是直接跳到最后几页，这个时候就不存在那个问题。
如果非要精准控制这个偏差就需要记住区间，或者用其他方式来实现了，比如全量查询表、sharding 索引表、最大下单 tps 值之类的，用来辅助计算。
（可以利用数据同步中间件建立单表多级索引、多表多维度索引来辅助计算。我们使用到的数据同步中间件有 datax、yugong、otter、canal 可以解决全量、增量同步问题）。
分表有多种方式，mod、rang、presharding、自定义路由，每种方式都有一定的侧重。
我们主要使用 mod + presharding 的方式，这种方式带来的最大的一个问题就是后期的节点变动数据迁移问题，可以通过参考一致性 hash 算法的虚拟节点来解决。
数据表拆分和 cache sharding 有一些区别，cache 能接受 cache miss ，通过被动缓存的方式可以维护起 cache 数据。但是数据库不存在 select miss 这种场景。
在 cache sharding 场景下一致性 hash 可以用来消除减少、增加 sharding node 时相邻分片压力问题。 但是数据库一旦出现数据迁移一定是不能接受数据查询不出来的。所以我们为了将来数据的平滑迁移，做了一个 虚拟节点 + 真实节点 mapping 。
physics node : node 1 node 2 node 3 node 4
virtual node : node 1 node 2 node 3.....node 20node mapping :
virtual node 1 ~ node 5 {physics node 1}
virtual node 6 ~ node 10 {physics node 2}
virtual node 11 ~ node 15 {physics node 3}
virtual node 16 ~ node 20 {physics node 4}为了减少将来迁移数据时 rehash 的成本和延迟的开销，将 hash 后的值保存在表里，将来迁移直接查询出来快速导入。
在我们熟悉的 hashmap 里，为了减少冲突和提供一定的性能将 hash 桶的大小设置成 2 的 n 次方，然后采用 hash&(legnth-1) 位与的方式计算，这样主要是大师们发现 2 的 n 次方的二进制除了高位是 0 之外所有地位都是 1，通过位与可以快速反转二进制然后地位加 1 就是最终的值。
我们在做数据库 sharding 的时候不需要参考这一原则，这一原则主要是为了程序内部 hash 表使用，外部我们本来就是要 hash mod 确定 sharding node 。
通过 mod 取模的方式会出现不均匀问题，在此基础上可以做个 自定义奇偶路由，这样可以均匀两边的数据。
1.在现有项目中集成 sharding-JDBC 有一些小问题，sharding-jdbc 不支持批量插入，如果项目中已经使用了大量的批量插入语句就需要改造，或者使用 辅助hash计算物理表名，在批量插入。
2.原有项目数据层使用 Druid + MyBatis，集成了 sharding-JDBC 之后 sharding-JDBC包装了 Druid ，所以一些 sharding-JDBC 不支持的sql语句基本就过不去了。
3.使用 springboot 集成 sharding-JDBC 的时候，在bean加载的时候我需要设置 IncrementIdGenerator ，但是出现classloader问题。
IncrementIdGenerator incrementIdGenerator = this.getIncrementIdGenerator(dataSource);
ShardingRule shardingRule = shardingRuleConfiguration.build(dataSourceMap);
((IdGenerator) shardingRule.getDefaultKeyGenerator()).setIncrementIdGenerator(incrementIdGenerator);private IncrementIdGenerator getIncrementIdGenerator(DataSource druidDataSource) {
...
}后来发现 springboot的类加载器使用的是 restartclassloader，所以导致转换一直失败。只要去掉 spring-boot-devtools package即可，restartclassloader 是为了热启动。
4.dao.xml 逆向工程问题，我们使用的很多数据库表mybatis生成工具生成的时候都是物理表名，一旦我们使用了sharding-JDCB之后都是用的逻辑表名，所以生成工具需要提供选项来设置逻辑表名。
5.为 mybatis 提供的 SqlSessionFactory 需要在Druid的基础上用shading-JDCB包装下。
6.sharding-JDBC DefaultkeyGenerator 默认采用是 snowflake 算法，但是我们不能直接用我们需要根据 datacenterid-workerid 自己配合zookeeper来设置 workerId 段。
（snowflake workId 10 bit 十进制 1023，dataCenterId 5 bit 十进制 31 、WorkId 5 bit 十进制 31）
7.由于我们使用的是 mysql com.mysql.jdbc.ReplicationDriver 自带的实现读写分离，所以处理读写分离会方便很多。如果不是使用的这种就需要手动设置 Datasource Hint 来处理。
8.在使用 mybatis dao mapper 的时候需要多份逻辑表，因为有些数据源数据表是不需要走sharding的，自定义shardingStragety 来处理分支逻辑。
9 全局id几种方法
9.1 如果使用 zookeeper 来做分布式ID，就要注意 session expired 可能会存在重复 workid 问题，加锁或者接受一定程度的并行（有序列号保证一段时间空间）。
9.2.采用集中发号器服务，在主DB中采用预生成表+incrment 插件（经典取号器实现，innodb 存储引擎中的 TRX_SYS_TRX_ID_STORE 事务号也是这种方式）
9.3.定长发号器、业务规则发号器，这种需要业务上下文的发号器实现都需要预先配置，然后每次请求带上获取上下文来说明获取业务类型
10.在项目中有些地方使用了自增id排序，数据表拆分之后就需要进行改造，因为ID大小顺序已经不存在了。根据数据的最新排序时使用了id排序需要改造成用时间字段排序。
作者：王清培 （沪江集团资深JAVA架构师）标签：datax 查询 virt 分布式系统 pack generator parameter 方案 压力
原文地址：http://blog.51cto.com/wangqingpei557/2148111
