经过一步步建立sharded cluster来认识mongodb

　　mongodb是目前使用很是普遍的nosql(not only sql)之一，在db engines上排名很是靠前，下图是5月份的排名：

　　

　　能够看到前面四个都是传统的关系型数据库，而mongodb在nosql中拔得头筹。本文会简单介绍mongodb的一些特性，而后经过在Linux环境下一步步搭建sharded cluster来学习mongodb。本文实验的mongodb是mongodb3.0，可能与最新的版本（mongodb3.4）在细节之处略有差别。

 

Mongodb特性

 　　官方一句话就能归纳Mongodb的特色：

　　MongoDB is an open-source document database that provides high performance, high availability, and automatic scaling.

　　开源、基于文档（document oriented）、高性能、高可用、自动伸缩。

　　开源：

　　这个好处就不用多说了，GitHub上有源码。

　　 面向文档：

　　文档（document）在不少编程语言都有相似的数据结构，各类table、map、dict，不再用使用DAO（data access object)。好比在python中，document与dict对应，array与list对应。

      document也支持嵌套的document和array，这样的话也能部分解决关联查询（固然，虽然把相关信息放在一个嵌套的document下降了关联查询的开销，但在某些状况不得不须要关联查询的时候仍是有点头疼）

　　因为基于document，因此就schema free（模式自由）啦，使用关系型数据的同窗都知道，线上修改表结构是多么麻烦的一件事情。但在mongodb中，增该删一个字段太容易了，这个也是最后开发人员喜欢的一点，好比游戏服务器，玩家的持久化数据会不停的变化，每次更新都会增长一些功能，也就回增长一些须要持久化的字段，用mongodb就很合适。

　　

　　高性能：

　　支持嵌套的document，在关系型数据库中须要联合查询的耗时操做，在mongodb中一条查询就能搞定

　　丰富的索引支持，并且索引还支持嵌套文档和数组

　　在使用了sharding机制的状况下，读写操做能够路由到不一样的shard，提到了集群的并发性能

 

　　高可用：

　　要想mongodb高可用，那么就得使用mongodb的复制机制：replica set

　　replica set经过异步复制（Asynchronous replication）和自动Failover来保证可用性

　　后面会专门介绍replica set

 

　　自动扩展（水平拆分）：

　　在关系型数据中，当单个表数据量过大的时候，通常会经过垂直分表或者水平分表的方式来提到数据库吞吐量。在mongodb中，sharding是其核心功能之一，提供了自动的水平扩展（horizontal scalability）来对数据量比较大的集合进行拆分，sharding将同一个集合的不一样子集数据放在不一样的机器上，当应用程序选择好适当的sharding key，能够将读写操做路由到某一个shard上，大大提升了集群吞吐性能。

　　后面会专门介绍sharding cluster的组成

 

预备知识

　　本文并不涵盖mongodb的基础知识，可是为了后面介绍sharding知识，以及搭建sharded cluster，在这里介绍一下 _id这个特殊的字段（field）

　　默认状况下，mongodb会在集合的_id字段上建立unique index。若是没有在持久化的文档中包含_id，那么mongodb会自动添加这个字段，其value是一个ObjectId。

　　mongodb官方建议_id使用ObjectId、或者天然惟一标示（unique identifier）、说着自增的数字、或者UUID。_id在sharding相关的CRUD中有特殊性，具体使用的时候能够参加文档。

　　另外，为了后文描述方便，这里声明几个在mongodb中的概念

　　DB：与mysql的DB对应

　　collection、集合： 与mysql的table对应

　　document：与mysql的record对应

replica set

　　MongoDB中经过 replica set（复制集）来提供高可用性：冗余与自动failover。

　　复制集是说将 同一份数据的多分拷贝放在不一样的机器（甚至不一样的数据中心）来提升容错。一个典型的replica set由一组mongod实例组成，其中有且仅有一个节点提供写操做，称之为primary，primary也是默认的读节点。同时，replica set中可包含一个到多个secondary节点，secondary节点只提供读操做。以下图所示：

　　

 

　　应用程序经过驱动与Primary链接，全部的写操做都在Primary上进行，同时primary会将这些操做写到oplog（operation log）中，secondary经过异步复制oplog，而后在本地数据集上执行oplog中的操做，这样就达到了数据的一致性。从这里能够看到，虽然secondary和primary维护的上同一份数据，可是其变动是要迟于primary的。

　　若是应用程序对数据的实时性要求不过高，好比评论数据、聚合数据等，那么能够从secondary读取，这样能够作到读写分离和高并发。若是拷贝放在不一样的数据中心，能更好的提供数据局部性（data locality）和分布式服务的可用性。

 

　　咱们看到，若是一个Secondary不能正常工做了（多是进程crash、物理损坏、网络故障），对应用程序来讲影响并不大。可是若是primary不能工做了呢？这个时候mongodb的automatic failover就开始发挥做用了。

　　在replica set中的所用mongod节点之间都会有心跳（heartbeat）存在，若是超过必定时间其余节点没有收到primary的心跳，那么就认为primary挂掉了。 可被选举的secondary会投票选举出新的primary。整个过程以下所示：

 

　　自动的failover 虽然保证了mongodb的高可用性，可是在primary到secondary的切换过程当中，这一段时间，mongodb是没法提供写操做的。表现就是对于应用程序的数据库操做请求会返回一些错误，这个时候应用程序须要识别这些错误，而后作重试。

　　

　　除了Primary和Secondary，在replica set中还能够存在存在另一种节点：Arbiter。Arbiter与Secondary节点的区别在于，Arbiter不持久化数据（do not bearing data),  天然也不可能在Primary挂掉的时候被选举。Arbiter的做用在于投票：为了选出新的primary，secondary投票规则是少数服从多数，若是replica set中的节点数目是偶数，那么就可能出现“平局”的状况，因此加入一个Arbiter就能够以最小的代价解决这个问题。

 

 　　Arbiter不持久化数据，因此占用的磁盘空间也不多，对硬件的要求也不高。官方建议，Arbiter不要和primary或者secondary放在同一个物理主机上。

　　在后面的演示中，也会在replica set中加入一个Arbiter，减小磁盘占用。

sharded cluster

　　所谓sharding就是将同一个集合的不一样子集分发存储到不一样的机器（shard）上，Mongodb使用sharding机制来支持超大数据量，将不一样的CRUD路由到不一样的机器上执行，提到了数据库的吞吐性能。因而可知，sharding是很是常见的scale out方法。

　　

 

　　如上图所示，一个集合（Collection1）有1T的数据，本来放在一个单独的数据库中，经过sharding，将这个集合的数据放在四个独立的shard中，每个shard存储这个集合256G的数据。每一个shard物理上是独立的数据库，但逻辑上共同组成一个数据库。

　　一个sharded cluster由一下三部分组成：config server，shards，router。如图所示：

　　

 

　　shards：

　　存储数据，能够是单个的mongod，也能够是replica set。在生产环境中，为了提升高可用性，都会使用replica set。存储在mongod上的数据以chunk为基本单位，默认的大小为64M，后面会介绍shard上数据的分裂（split）与迁移（migration）

　　config server：

　　存储集群的元数据（metadata），即数据的哪一部分放在哪个shard上，router将会利用这些元数据将请求分发到对应的shards上，shards上chunk的迁移也是config server来控制的。

　　router:

　　mongos实例，在一个集群中直接为应用程序提供服务，利用config server上的元数据来制定最佳的查询计划。

 

　　数据分割（data partition）：

　　从前文知道，MongoDB在collection这个级别进行数据的切块，称之为sharding。块的最小粒度是chunk，其大小（chunkSize）默认为64M。

　　当一个集合的数据量超过chunkSize的时候，就会被拆分红两个chunk，这个过程称为splitting。那么按照什么原则将一个chunk上的数据拆分红两个chunk，这就是Sharding key的做用，Sharding key是被索引的字段，经过sharding key，就能够把数据均分到两个chunk，每个document在哪个chunk上，这就是元数据信息。元数据信息存放在config server上，方便router使用。

　　若是sharding cluster中有多个shard，那么不一样shard上的chunk数目多是不一致的，这个时候会有一个后台进程（balancer）来迁移（migrate）chunk，从chunk数目最多的shard迁移到chunk数目最少的chunk，直到达到均衡的状态。迁移的过程对应用程序来讲是透明的。

　　以下图所示，迁移以前ShardA ShardB上都有3个chunk，而Shard C上只有一个Chunk。经过从ShardB上迁移一个chunk到ShardC，就达到了一个均衡的状态。　　

　　splitting和migration 的目的是为了让数据在shards之间均匀分布，其根本目标是为了将对数据的CRUD操做均衡地分发到各个shard，提升集群的并发性能。

　　

Sharded cluster搭建

　　声明，本章节只是演示Sharded Cluster的搭建过程，与生产环境仍是有较大差别，不过我也会在文中尽可能指出这些差别。首先须要注意的是，本文的演示不涉及到鉴权(--auth)，但在生产环境中鉴权是很是重要的，相信你们都还记得春节期间Mongodb被劫持、被攻击的事件。

　　前文已经提到，一个典型的Sharded Cluster包括router(mongos)、config server和shards，其中每一个shard均可以是单点(standalone)或者复制集(replica set）。 接下来的演示包括一个router， 三个config server，两个shard。每个shard都是有一个primary、一个secondary和一个arbiter组成的replica set。

　　在开始以前，首先预约义好全部须要用到的变量，以下所示：

 1 #!/bin/bash
 2 export BIN_HOME=/usr/local/mongodb/bin
 3 export DB_PATH=/home/mongo_db/data
 4 export LOG_PATH=/home/mongo_db/log
 5 
 6 LOCAL=127.0.0.1
 7 
 8 #config rs
 9 export RS1_1_DB_PATH=$DB_PATH/rs1_1
10 export RS1_2_DB_PATH=$DB_PATH/rs1_2
11 export RS1_3_DB_PATH=$DB_PATH/rs1_3
12 export RS2_1_DB_PATH=$DB_PATH/rs2_1
13 export RS2_2_DB_PATH=$DB_PATH/rs2_2
14 export RS2_3_DB_PATH=$DB_PATH/rs2_3
15 
16 export RS1_1_DB_LOG=$LOG_PATH/rs1_1.log
17 export RS1_2_DB_LOG=$LOG_PATH/rs1_2.log
18 export RS1_3_DB_LOG=$LOG_PATH/rs1_3.log
19 export RS2_1_DB_LOG=$LOG_PATH/rs2_1.log
20 export RS2_2_DB_LOG=$LOG_PATH/rs2_2.log
21 export RS2_3_DB_LOG=$LOG_PATH/rs2_3.log
22 
23 export RS1_1_PORT=27018
24 export RS1_2_PORT=27019
25 export RS1_3_PORT=27020
26 export RS2_1_PORT=27021
27 export RS2_2_PORT=27022
28 export RS2_3_PORT=27023
29 
30 export RS1=rs1
31 export RS2=rs2
32 
33 #config config_server
34 export CONF1_DB_PATH=$DB_PATH/db_conf1
35 export CONF2_DB_PATH=$DB_PATH/db_conf2
36 export CONF3_DB_PATH=$DB_PATH/db_conf3
37 
38 export CONF1_DB_LOG=$LOG_PATH/conf1.log
39 export CONF2_DB_LOG=$LOG_PATH/conf2.log
40 export CONF3_DB_LOG=$LOG_PATH/conf3.log
41 
42 export CONF1_PORT=40000
43 export CONF2_PORT=40001
44 export CONF3_PORT=40002
45 
46 export CONF1_HOST=$LOCAL:$CONF1_PORT
47 export CONF2_HOST=$LOCAL:$CONF2_PORT
48 export CONF3_HOST=$LOCAL:$CONF3_PORT
49 
50 #config route_server
51 export ROUTE_DB_LOG=$LOG_PATH/route.log
52 
53 export ROUTE_PORT=27017

 　　能够在会话窗口中将这些命令执行一遍，不过更好的方式是将其保存在一个文件中（如mongodb_define.sh），而后执行这个文件就好了：source mongodb_define.sh

启动shards（replica set）

　　在这一部分，会建立两个replica set，分别是rs1， rs2。每一个replica set包含三个节点，且其中一个是arbiter。因为两个replica set建立过程没什么区别，所以以rs1为例。关于replica set的搭建，可参见mongodb doc中 deploy-replica-set-for-testing部分，讲得比较清楚。

　　 step1: 首先得建立好存放数据的目录：

　　mkdir -p $RS1_1_DB_PATH

　　mkdir -p $RS1_2_DB_PATH

　　mkdir -p $RS1_3_DB_PATH

　　PS: -p means "no error if existing, make parent directories as needed"

 

　　 step2: 启动组成rs1的三个mongod　

　　$BIN_HOME/mongod --port $RS1_1_PORT --dbpath $RS1_1_DB_PATH --fork --logpath $RS1_1_DB_LOG --replSet $RS1 --smallfiles --nojournal
　　$BIN_HOME/mongod --port $RS1_2_PORT --dbpath $RS1_2_DB_PATH --fork --logpath $RS1_2_DB_LOG --replSet $RS1 --smallfiles --nojournal
　　$BIN_HOME/mongod --port $RS1_3_PORT --dbpath $RS1_3_DB_PATH --fork --logpath $RS1_3_DB_LOG --replSet $RS1 --smallfiles --nojournal

　　关于mongod的启动选项，能够经过mongod --help查看，在上面的命令行中，--replSet 指定了replica set的名字， --smallfiles 声明使用更小的默认文件， --nojournal代表不开启journaling机制。注意，在这个地方不开启journaling是由于实验环境磁盘空间有限，而全部的mongod实例都在这个机器上，在生成环境中，必定要开始journaling，这个是mongodb durability的保证。

　　

　　 step3：初始化复制集rs1

　　在这一步，须要经过mongdb的客户端mongo链接到复制集的任何一个节点，对复制集初始化，这里链接到RS1_1（端口为27018）：

　　mongo --port $RS1_1_PORT

　　先来看一下如今复制集的状态（PS：下面全部以 > 开头的命令行都表示是在mongo这个交互式客户端输入的指令）

　　> rs.status()

　　

　　能够看到这个复制集尚未初始化

　　>config = {

　　　　_id : "rs1",
　　　　members : [
　　　　　　{_id : 0, host : "127.0.0.1:27018"},
　　　　　　{_id : 1, host : "127.0.0.1:27019"},
　　　　　　{_id : 2, host : "127.0.0.1:27020", arbiterOnly: true},
　　　　]
　　}

　　>rs.initiate(config)

　　

　　从config和运行后的复制集状态均可以看到，RS1_3（127.0.0.1:27020）这个mongod为一个Arbiter，即只参与投票，不持久化数据。另外RS1_1为Primary， RS1_2为Secondary。

　　到此为止，复制集rs1就启动好了。

 

 

　　关于s2的启动，下面也给出全部命令。方便读者实践

　　mkdir -p $RS2_1_DB_PATH
　　mkdir -p $RS2_2_DB_PATH
　　mkdir -p $RS2_3_DB_PATH

　　$BIN_HOME/mongod --port $RS1_1_PORT --dbpath $RS1_1_DB_PATH --fork --logpath $RS1_1_DB_LOG --replSet $RS1 --smallfiles --nojournal
　　$BIN_HOME/mongod --port $RS1_2_PORT --dbpath $RS1_2_DB_PATH --fork --logpath $RS1_2_DB_LOG --replSet $RS1 --smallfiles --nojournal
　　$BIN_HOME/mongod --port $RS1_3_PORT --dbpath $RS1_3_DB_PATH --fork --logpath $RS1_3_DB_LOG --replSet $RS1 --smallfiles --nojournal

　　mongo --port $RS2_1_PORT
　　>config = {
　　_id : "rs2",
　　members : [
　　{_id : 0, host : "127.0.0.1:27021"},
　　{_id : 1, host : "127.0.0.1:27022"},
　　{_id : 2, host : "127.0.0.1:27023", arbiterOnly: true},
　　]
　　}
　　>rs.initiate(config)

启动config servers

　　mongodb官方建议config server须要三个mongod实例组成，每个mongod最好部署在不一样的物理机器上。这个三个mongod并非复制集的关系，

　　step1：建立db目录

　　mkdir -p $CONF1_DB_PATH
　　mkdir -p $CONF2_DB_PATH
　　mkdir -p $CONF3_DB_PATH

　　step2：启动三个mongod实例：

　　$BIN_HOME/mongod --port $CONF1_PORT --dbpath $CONF1_DB_PATH --fork --logpath $CONF1_DB_LOG --configsvr --smallfiles --nojournal
　　$BIN_HOME/mongod --port $CONF2_PORT --dbpath $CONF2_DB_PATH --fork --logpath $CONF2_DB_LOG --configsvr --smallfiles --nojournal
　　$BIN_HOME/mongod --port $CONF3_PORT --dbpath $CONF3_DB_PATH --fork --logpath $CONF3_DB_LOG --configsvr --smallfiles --nojournal

　　一样启动参数中nojournal只是为了节省存储空间，在生产环境中必定要使用journaling。与建立replica set时mongod的启动不一样的是，这里有一个configsvr 选项，代表这些节点都是做为config server存在。

　　再启动这三个mongod以后，不会有相似replica set那样讲三个mongod绑定之类的操做，也说明了config server之间是相互独立的

 

启动router

　　在Sharded Cluster中，router（mongos）是应用程序链接的对象，一切对mongodb的操做都经过router来路由

　　 step1：启动mongos

 

　　 $BIN_HOME/mongos --port $ROUTE_PORT --configdb $CONF1_HOST,$CONF2_HOST,$CONF3_HOST --fork --logpath $ROUTE_DB_LOG --chunkSize 32

　　注意这里的可执行程序是mongos，而不是以前的mongod，关于参数，也是能够经过mongos --help查看的。在上面的命令中，--configdb选项指定了三个config server，--chunkSize指定了chunk的大小，单位为M。关于chunksize，默认是64M，虽然能够在初次启动的时候指定chunksize，但mongodb官方推荐按照 如下方式修改。 在本文中将chunkSize改小的目的，是为了之后实验的时候更方便观察数据的拆分和迁移。

　　chunkSize事实上会持久化到config.setting中，链接到mongos可查看：

　　mongo --port $ROUTE_PORT 

　　

　　在上面截图蓝色框中能够看出，如今尚未任何shard的信息，缘由是到如今为止，config servers与replica set尚未任何关系

 

　　 step2：将在前面建立的两个replica set（rs1 rs2）加入到Sharded Cluster中

　　 mongo --port $ROUTE_PORT 

　　mongos> sh.addShard('rs1/127.0.0.1:27018')
　　mongos> sh.addShard('rs2/127.0.0.1:27021')

　　PS：为何须要在rs1后面指定一个mongod的ip port，这个是用来找到对应的mongod，继而找到相应rs

　　再次查看结果：

　　

　　能够看到已经添加了两个shard，每个都是一个replica set。有意思的是Arbiter（好比RS1_3）并无显示在查询结果中，可能的缘由是Arbiter并不持久化数据，显示在这里也没有什么意义。

 

　　到此为止，整个Sharded Cluster就算搭建好了，可是还未进入真正使用阶段，要发挥Sharded Cluster的做用，咱们得指定哪些collection能够被sharding，以及如何sharding

建立sharding key

　　为了演示，咱们假设添加一个db叫test_db, 其中有两个collection，一个是须要sharding的，叫sharded_col；另外一个暂时不用sharding，叫non_sharded_col， 固然以后也能够增长新的集合，或者把原来没有sharding的集合改为sharding。

　　一下操做都须要登陆到router进行： mongo --port $ROUTE_PORT 

　　step1：首先得告知mongodb test_db这个数据库支持sharding

　　mongos> sh.enableSharding("test_db")
　　{ "ok" : 1 }

　　这个时候能够查看数据库的状态，注意，是在config这个db下面的databases集合

　　mongos> use config

　　mongos> db.databases.find()

　　{ "_id" : "admin", "partitioned" : false, "primary" : "config" }

　　{ "_id" : "test_db", "partitioned" : true,  "primary" : "rs1" }

　　从查询结果能够看到，test_db是支持sharding的（"partitioned" : true）。另外上面加粗部分primary： rs1，这个primary与replica set里面的primary不是一回事，这里的primary是primary shard的意思，咱们知道即便在sharded cluster环境中，一些数据量比较小的db也是无需分片的，这些db默认就存放在primary shard上面

 

　　 step2：为须要的collection（即这里的sharded_col）指定sharding key

　　前面已经提到了sharding key的做用，关于sharding key的选择，是一个比较复杂的问题，sharding key对索引，对CRUD语句的操做都有诸多限制，这一部分之后再细讲，在这里默认使用_id作sharding key（_id是mongodb默认为每一个集合增长的索引）

　　mongos> sh.shardCollection('test_db.sharded_col', {'_id': 1})

　　接下来看看整个sharded cluster的状态：

　　

　　sh.status()反应的内容事实上也是来自config整个数据库的内容，只不过作了必定程度的整合。从上面能够看到，有两个shard，rs1, rs2;test_db容许sharding，test_db.sharded_col整个collection的sharding key为{"_id": 1}，且目前只有一个chunk在rs1整个shard上。

总结：

　　到目前为止，咱们已经搭建了一个有三个config server，两个shard的sharded cluster，其中每个shard包含三个节点的replica set，且都包含一个Arbiter。咱们能够查看一下刚建立好以后各个mongod实例持久化的数据大小：

　　

　　能够看到，两个Arbiter（rs1_3, rs2_3）所占的空间要小得多。

　　对于应用程序来讲，集群（sharded cluster）和单点（standalone）是有必定差别的，若是须要发挥sharded cluster高性能、高可用的特色，须要根据应用场景精心选择好sharding key，而sharding key的选择跟索引的创建以及CRUD语句息息相关，这一部分之后再聊。对于目前搭建的这个实例，简单测试的话，往sharded_col插入足够多条document就能看到chunks的拆分和迁移。

 

references：

db engines

the-id-field

replication-introduction

deploy-replica-set-for-testing

deploy-shard-cluster