Mongodb 源码分析–Replication之主从模式–Master

不念不忘少年蓝@ 2022-06-18 06:29 86阅读 0赞

# [Mongodb 源码分析–Replication之主从模式–Master][Mongodb _Replication_Master] #

转自： [http://www.wdcode.org/archives/488.html][Mongodb _Replication_Master]

mongodb中提供了复制(Replication)机制，通过该机制可以帮助我们很容易实现读写分离方案，并支持灾难恢复（服务器断电）等意外情况下 的数据安全。

在老版本(1.6)中，Mongo提供了两种方式的复制：master-slave及replica pair模式（注:mongodb最新支持的replset复制集方式可看成是pair的升级版，它解决pair只能在两个结点间同步的限制，支持多个结 点同步且支持主从宕机时的自动切换, 在1.6版以后提供）。  
![0_13186640375bXO.gif][]

利用前者，我们可以实现读写分离（主从复制模式），后者则支持当主服务器断电情况下的集群中其它slave自动接管，并升级为主服务器。 并且如果后来的也出错了，那么master状态将会转回给第一个服务器（之前宕机但后来又恢复运行的服务器）。

同时mongodb支持使用安全认证（enable）。不管哪种replicate方式，只要在master/slave中创建一个能为各个 database认识的用户名/密码即可。其认证过程如下：

slave先在local.system.users里查找 一个名为”repl”的用户，找到后用它去认证master。如果”repl”用户没有找到，则使用local.system.users中的第一个用户 去认证。local数据库和admin数据库一样，local中的用户可以访问整个db server。

下面介绍分别介绍一下这两种复制的配置方式：

**Master-Slave(主从)模 式:**  
一个server可以同时为master和slave。一个slave可以有多个master（不推荐，可能会产生不可预期的结果）。

配置选项：  
–**master** 以主服务器方式启动  
–**slave** 以从服务器方式启动  
–**autoresync** ：自动重新sync，因为该操作会copy 主服务器上的所有document，比较耗时，在10分钟内最多只会进行一次。  
–**oplogSize** ：指定master上用于存放更改的数据量，如果不指定，在32位机上最少为 50M，在64位机上最少为 1G，最大为磁盘空间的5%。  
–**source** 主服务器地址（与–slave组合使用）  
–**only** 仅限于同步指定数据库（下面示例为test库）  
–**slavedelay** 同步延时

下面是本人在本地为了测试方便所使用的配置参数  
Master: IP->10.0.1.103

mongod –dbpath=d:\\mongodb\\db –master –oplogSize 64

Slave: IP->10.0.4.210

mongod –dbpath=d:\\mongodb\\db –slave –source 10.0.1.103:27017 –only test –slavedelay 100

补充：受限的master-master复制，这种模式对插入、查询及根据\_id进行的删除操作都是安全的。但对同一对象的并发更新无法进行。 Mongo 不支持完全的master-master复制，通常情况下不推荐使用master-master模式，但在一些特定的情况下master- master也可用。master-master也只支持最终一致性。配置master-master只需运行mongod时同时加上–master选 项和 –slave选项。如下：  
mongod –dbpath=d:\\mongodb\\db –port 27017 –master –slave –source localhost:27018  
mongod –dbpath=d:\\mongodb\\db –port 27018 –master –slave –source localhost:27017

**Replica pairs模式**  
以这种方式启动后，数据库会自动协商谁是master谁是slave。一旦一个数据库服务器断电，另一个会自动接管，并从那一刻起起为master。万一 另一个将来也出错了，那么master状态将会转回给第一个服务器。以这种复制方式启动mongod的命令如下：  
配置选项：  
mongod –pairwith <remoteserver> –arbiter <arbiterserver>  
–**pairwith** : remoteserver是pair里的另一个server  
–**arbiter** : arbiterserver是一个起仲裁作用的Mongo数据库，用来协商pair中哪一个是master。arbiter运行在第三个机器上，利用“平 分决胜制”决定在pair中的两台机器不能联系上对方时让哪一个做master，一般是能同arbiter通话的那台机器做master。如果不加 –arbiter选项，出现网络问题时两台机器都作为master。  
注：可使用db.$cmd.findOne(\{ismaster:1\})可以检查当前哪一个database是master。

另外这种模式下的两台机器只能满足最终一致性。当replica pair中的一台机器完全挂掉时，需要用一台新的来代替。如(n1, n2)中的n2挂掉，这时用n3来代替n2。步骤如下：  
1. 告诉n1用n3来代替n2：db.$cmd.findOne(\{replacepeer:1\});  
2. 重启n1让它同n3对话：mongod –pairwith n3 –arbiter <arbiterserver>  
3. 启动n3：mongod –pairwith n1 –arbiter <arbiterserver>。  
在n3的数据没有同步到n1前n3还不能做master，这个过程长短由数据量的多少决定。

了解了复制模式之后，还有一个问题需要介绍一下，不是就是本文中mongodb使用cap collection来存储操作日志，并进而使用日志来复制(同步)结点间的数据，其中由主结点保存的操作的记录叫做oplog(operation log的简称)。

Oplog存在一个叫local的特殊数据库中，在oplog.$main集合。Oplog中的每一个文档表示一个在主结点上执行的操作。文档主要包括4 块内容，如下：

Ts:操作的时间戳。时间戳类型是一个用来跟踪操作是何时执行的一 种内部类型。它由4字节的时间戳和四字节的增量计数器组成。  
Op:执行的操作的类型，大小为1字节。（例如，“i”代表insert,”u”:update, ”d”:delete, ”n”:none无操作 等）  
Ns：执行操作的命名空间（集合名）  
O：执行操作的文档。对于插入，这是将要插入的文档。

另外这种日志只保存会“改变数据库状态”的操作。查询操作不会记录在oplog中。

好了，了解这些知识之后，我们就来开始看一下如何调试master-slave模式的源码，首先要在vs2010中打开mongod项目，并将启动参数中 设置如下：  
–master –oplogSize 64 （master IP为10.0.1.103）

如下图:

![0_13186640446sAc.gif][]

之后编译该项目，启动该主服务结点，如下：

![0_1318664059P9oN.gif][]  
接着我们可以在本地或另外一台机器上启动一个slave结点：

mongod –dbpath=d:\\mongodb\\db –slave –source 10.0.1.103:27017 –only test –slavedelay 100

下面介绍一下master（主服务端）的代码执行流程。首先我们打开instance.cpp文件，找到下面方法：

[view plain][] [copy to clipboard][view plain] [print][view plain] [?][view plain]

1.  //instance.cpp
2.  // Returns false when request includes ’end’
3.  void assembleResponse( Message &m, DbResponse &dbresponse, const SockAddr &client ) \{
4.  ……
5.  if ( op == dbQuery ) \{
6.  if ( handlePossibleShardedMessage( m , &dbresponse ) )
7.  return;
8.  receivedQuery(c , dbresponse, m );
9.  \}
10. //服务端(master) 收到message执行相关查询操作
11. else if ( op == dbGetMore ) \{
12. if ( ! receivedGetMore(dbresponse, m, currentOp) )
13. log = true;
14. \}
15. …..
16. \}

看过本系列开头那几篇BLOG的朋友，会看出上面方法其实在mongodb的crud操作中都会执行到，更多内容可以参见[这篇 BLOG][BLOG] ，这里不再赘述。

当slave 从结点发送同步复制请求时，master会执行上面的dbGetMore操作，从主库中的oplog中获取相应日志并返回给slave结点，下面是 receivedGetMore()方法的具体实现：

[view plain][] [copy to clipboard][view plain] [print][view plain] [?][view plain]

1.  //instance.cpp
2.  bool receivedGetMore(DbResponse& dbresponse, Message& m, CurOp& curop ) \{
3.  StringBuilder& ss = curop.debug().str;
4.  bool ok = true;
5.  //参见：Mongodb源码分析–消息(message)中的 查询更多 (document)消息结构相关内容
6.  //http://www.cnblogs.com/daizhj/archive/2011/04/02/2003335.html
7.  DbMessage d(m);
8.  //完整的集合名称，形如:”dbname.collectionname”
9.  const char \*ns = d.getns();
10. //返回的document数
11. int ntoreturn = d.pullInt();
12. //在REPLY消息中的Cursor标识符，其必须来自于数据库
13. long long cursorid = d.pullInt64();
14. ss << ns << ” cid:” << cursorid;
15. if( ntoreturn )
16. ss << ” ntoreturn:” << ntoreturn;
17. time\_t start = 0;
18. int pass = 0;
19. bool exhaust = false;
20. QueryResult\* msgdata;//查询结果
21. while( 1 ) \{
22. try \{
23. readlock lk;
24. Client::Context ctx(ns);
25. //执行GetMore查询
26. msgdata = processGetMore(ns, ntoreturn, cursorid, curop, pass, exhaust);
27. \}
28. catch ( GetMoreWaitException& ) \{
29. exhaust = false;
30. massert(13073, ”shutting down”, !inShutdown() );
31. if( pass == 0 ) \{
32. start = time(0);
33. \}
34. else \{
35. if( time(0) - start >= 4 ) \{
36. // after about 4 seconds, return. this is a sanity check. pass stops at 1000 normally
37. // for DEV this helps and also if sleep is highly inaccurate on a platform. we want to
38. // return occasionally so slave can checkpoint.
39. pass = 10000;
40. \}
41. \}
42. pass++;
43. DEV
44. sleepmillis(20);
45. else
46. sleepmillis(2);
47. continue;
48. \}
49. catch ( AssertionException& e ) \{
50. exhaust = false;
51. ss << ” exception ” << e.toString();
52. msgdata = emptyMoreResult(cursorid);
53. ok = false;
54. \}
55. break;
56. \};
57. //将查询结果集绑定到message对象
58. Message \*resp = new Message();
59. resp->setData(msgdata, true);
60. ss << ” bytes:” << resp->header()->dataLen();
61. ss << ” nreturned:” << msgdata->nReturned;
62. //将上面的消息对象指针绑定到dbresponse
63. dbresponse.response = resp;
64. dbresponse.responseTo = m.header()->id;
65. if( exhaust ) \{
66. ss << ” exhaust ”;
67. dbresponse.exhaust = ns;
68. \}
69. return ok;

可以看出，通过对message的解析找出相应的cursorid，因为mongodb如果发现游标为tailable(类型)时，会cache该 cursor而不是关闭它，这主要是考虑到当下次slave请求来时，直接从cache中获取该cursor以提升效率并用它来作为继续获取后续 oplog操作信息。上面方法在执行结束处会将获取到的oplog结果封装到message中并返回。但其如何获取，就要分析下面方法了：

[view plain][] [copy to clipboard][view plain] [print][view plain] [?][view plain]

1.  //query.cpp
2.  QueryResult\* processGetMore(const char \*ns, int ntoreturn, long long cursorid , CurOp& curop, int pass, bool& exhaust ) \{
3.  exhaust = false;
4.  //在map<CursorId, ClientCursor\*>中查询相应游客信息
5.  ClientCursor::Pointer p(cursorid);
6.  //将结果返回（可能没找到）
7.  ClientCursor \*cc = p.c();
8.  int bufSize = 512;
9.  if ( cc ) \{
10. bufSize += sizeof( QueryResult );
11. bufSize += MaxBytesToReturnToClientAtOnce;
12. \}
13. //创建收集查询记录结果的buf对象
14. BufBuilder b( bufSize );
15. //跳过预留数据区间（QueryResult）
16. b.skip(sizeof(QueryResult));
17. int resultFlags = ResultFlag\_AwaitCapable;
18. int start = 0;
19. int n = 0;
20. //判断cc是否有效(如未找到则无效)
21. if ( !cc ) \{
22. log() << ”getMore: cursorid not found ” << ns << ” ” << cursorid << endl;
23. cursorid = 0;
24. resultFlags = ResultFlag\_CursorNotFound;
25. \}
26. else \{
27. //更新master结点local.slaves中的相应信息（包括lastop时间戳）
28. //注：主结点使用存储在local.slaves中的syncedTo来跟踪多少 slave是已经更新的。
29. if ( pass == 0 )
30. cc->updateSlaveLocation( curop );
31. int queryOptions = cc->queryOptions();
32. if( pass == 0 ) \{
33. StringBuilder& ss = curop.debug().str;
34. ss << ” getMore: ” << cc->query().toString() << ” ”;
35. \}
36. //获取相应cursor，以便while遍历
37. start = cc->pos();
38. Cursor \*c = cc->c();
39. c->checkLocation();
40. DiskLoc last;
41. scoped\_ptr<Projection::KeyOnly> keyFieldsOnly;
42. if ( cc->modifiedKeys() == false && cc->isMultiKey() == false && cc->fields )
43. keyFieldsOnly.reset( cc->fields->checkKey( cc->indexKeyPattern() ) );
44. //遍历cursor，找到并封装相应查询结果给buf对象
45. while ( 1 ) \{
46. if ( !c->ok() ) \{//到结尾
47. if ( c->tailable() ) \{//处理 tailable情况
48. //Tailable 表示在返回最后一条数据后，不要关闭当前 cursor。
49. //这是因为系统考虑到稍后你可以再次使用该cursor.
50. /\* when a tailable cursor hits ”EOF”, ok() goes false, and current() is null. however
51. advance() can still be retries as a reactivation attempt. when there is new data, it will
52. return true. that’s what we are doing here.
53. \*/
54. if ( c->advance() )
55. continue;
56. if( n == 0 && (queryOptions & QueryOption\_AwaitData) && pass < 1000 ) \{
57. throw GetMoreWaitException();
58. \}
59. break;
60. \}
61. //释放cursor资源关闭它（执行delete操作）
62. p.release();
63. bool ok = ClientCursor::erase(cursorid);
64. assert(ok);
65. cursorid = 0;
66. cc = 0;
67. break;
68. \}
69. // 如果是clone collection时，则不会匹配
70. // If match succeeds on index key, then attempt to match full document.
71. if ( c->matcher() && !c->matcher()->matches(c->currKey(), c->currLoc() ) ) \{
72. \}
73. /\*
74. TODO
75. else if ( \_chunkMatcher && ! \_chunkMatcher->belongsToMe( c->currKey(), c->currLoc() ) )\{
76. cout << ”TEMP skipping un-owned chunk: ” << c->current() << endl;
77. \}
78. \*/
79. else \{//值是否重复
80. if( c->getsetdup(c->currLoc()) ) \{
81. //out() << ” but it’s a dup \\n”;
82. \}
83. else \{//如匹配
84. last = c->currLoc();
85. n++;
86. //装填数据到buf中
87. if ( keyFieldsOnly ) \{
88. fillQueryResultFromObj(b, 0, keyFieldsOnly->hydrate( c->currKey() ) );
89. \}
90. else \{
91. BSONObj js = c->current();
92. // show disk loc should be part of the main query, not in an $or clause, so this should be ok
93. fillQueryResultFromObj(b, cc->fields.get(), js, ( cc->pq.get() && cc->pq->showDiskLoc() ? &last : 0));
94. \}
95. if ( ( ntoreturn && n >= ntoreturn ) || b.len() > MaxBytesToReturnToClientAtOnce ) \{
96. c->advance();
97. cc->incPos( n );
98. break;
99. \}
100. \}
101. \}
102. //指向下一条记录
103. c->advance();
104. if ( ! cc->yieldSometimes() ) \{
105. cc = 0;
106. break;
107. \}
108. \}
109. if ( cc ) \{
110. cc->updateLocation();
111. cc->mayUpgradeStorage();
112. //用last中的optime 更新\_slaveReadTill
113. cc->storeOpForSlave( last );
114. exhaust = cc->queryOptions() & QueryOption\_Exhaust;
115. \}
116. \}
117. //将buf中的信息绑定到查询结果集
118. QueryResult \*qr = (QueryResult \*) b.buf();
119. qr->len = b.len();
120. qr->setOperation(opReply);
121. qr->\_resultFlags() = resultFlags;
122. qr->cursorId = cursorid;
123. qr->startingFrom = start;
124. qr->nReturned = n;
125. b.decouple();
126. return qr;
127. \}

上面代码有些长，但其目的很明确，就是针对指定的cursor进行遍历。这里mongodb会为每个slave保存一个cursor，并且其在遍历完成后 将最后一条oplog的时间戳作为当前slave在local.slaves中的更新标识信息(syncedTo)，来标识当前slave的更新情 况。（注：首次同步时全部复制会执行copyDatabase，复制master db上的所有document）。该方法运行截图如下：

![0_131866406559Rr.gif][]

另外需要解释的是，master结点貌似并不会使用slave发来的syncedTo来过滤capped collection中的旧oplog(指小于syncedTo时间戳)的数据，而是使用tailable类型的cursor来解决如果持续获取后续新增 oplog操作信息。前者的主观臆测让我在源码中兜了一个圈子，因为我一直主观认为mongod会执行类似查询操作来过滤相应旧oplog的时间戳信息， 并将结果集返回给slave端。现在看来master只是不断返回后续添加到cap collection中oplog（有可能是out of sync的情况而引发slave地点执行resync操作），而最终的过滤判断操作完全交给了slave端。这一点会在我下一篇文章中有所介绍。

[Mongodb _Replication_Master]: http://www.wdcode.org/archives/488.html
[0_13186640375bXO.gif]: /images/20220618/fbbeb6e736074beba3638bc39628cc34.png
[0_13186640446sAc.gif]: /images/20220618/8ad87a743a2c45eb8461a6b5c05d623d.png
[0_1318664059P9oN.gif]: /images/20220618/8210dab7b4824cdcb5b171903afffef4.png
[view plain]: http://blog.csdn.net/daizhj/archive/2011/06/13/6540988.aspx#
[BLOG]: http://www.cnblogs.com/daizhj/archive/2011/03/18/1988288.html
[0_131866406559Rr.gif]: /images/20220618/84aa0f9a69de47e39176ebdaeb8b23ce.png