Mongodb源码分析之Mongos分析

MongoDB提供了auto-sharding 功能。因为其是auto-sharding，即mongodb通过mongos(一个自动分片模块，用于构建一个大规模的可扩展的数据库集群,这个集群可以并入动态增加的机器)自动建立一个水平扩展的数据库集群系统，将数据库分表存储在sharding的各个节点上。

我们提供的服务有：网站建设、网站设计、微信公众号开发、网站优化、网站认证、青白江ssl等。为1000多家企事业单位解决了网站和推广的问题。提供周到的售前咨询和贴心的售后服务，是有科学管理、有技术的青白江网站制作公司

一个mongodb集群包括一些shards(包括一些mongod进程)，mongos路由进程，一个或多个config服务器

下面是一些相关词汇说明:

Shards :每一个shard包括一个或多个服务和存储数据的mongod进程(mongod是MongoDB数据的核心进程)典型的每个shard开启多个服务来提高服务的可用性。这些服务/mongod进程在shard中组成一个复制集

Chunks:Chunk是一个来自特殊集合中的一个数据范围，(collection，minKey，maxKey)描叙一个chunk，它介于minKey和maxKey范围之间。例如chunks 的maxsize大小是100M，如果一个文件达到或超过这个范围时，会被切分到2个新的chunks中。当一个shard的数据过量时，chunks将会被迁移到其他的shards上。同样，chunks也可以迁移到其他的shards上

Config Servers : Config服务器存储着集群的metadata信息，包括每个服务器，每个shard的基本信息和chunk信息Config服务器主要存储的是chunk信息。每一个config服务器都复制了完整的chunk信息。

今天要介绍的源码主要是Mongos的主入口函数的执行流程，首先我们打开Mongos的项目(可通过打开源码db\db_10.sln加载所有项目)，如下图：

注：如果要调试mongos，需要设置一个mongod进程和一个Config Server,形如：

d:\mongodb>bin>mongod --dbpath d:\mongodb\db\ --port 27012

d:\mongodb>bin>mongod --configsvr --dbpath d:\mongodb\db\ --port 27022

然后在vs2010中配置相应的boost路径信息及启动参数信息，如下图：

#p#

下面开始正文。首先打开mongos项目中的server.cpp文件，找到下面方法：

 
 
 
 
      
  
  
  
  1. int main(int argc, char* argv[]) {  
    2.   
  
  
  
  2.     try {  
    3.   
  
  
  
  3.         return _main(argc, argv);  
    4.   
  
  
  
  4.     }  
    5.   
  
  
  
  5.     catch(DBException& e) {  
    6.   
  
  
  
  6.         cout << "uncaught exception in mongos main:" << endl;  
    7.   
  
  
  
  7.         cout << e.toString() << endl;  
    8.   
  
  
  
  8.     }  
    9.   
  
  
  
  9.     catch(std::exception& e) {  
    10.   
  
  
  
  10.         cout << "uncaught exception in mongos main:" << endl;  
    11.   
  
  
  
  11.         cout << e.what() << endl;  
    12.   
  
  
  
  12.     }  
    13.   
  
  
  
  13.     catch(...) {  
    14.   
  
  
  
  14.         cout << "uncaught exception in mongos main" << endl;  
    15.   
  
  
  
  15.     }  
    16.   
  
  
  
  16.     return 20;  
    17.   
  
  
  
  17. } 
    18.

该方法是mongos的主函数，代码很简，它主要是try方式执行_main方法，下面是_main的执行流程：

 
 
 
 
      
  
  
  
  1. int _main(int argc, char* argv[]) {  
    2.   
  
  
  
  2.     static StaticObserver staticObserver;  
    3.   
  
  
  
  3.     mongosCommand = argv[0];  
    4.   
  
  
  
  4.     //声明options信息描述对象  
    5.   
  
  
  
  5.     po::options_description options("General options");  
    6.   
  
  
  
  6.     po::options_description sharding_options("Sharding options");  
    7.   
  
  
  
  7.     po::options_description hidden("Hidden options");  
    8.   
  
  
  
  8.     po::positional_options_description positional;  
    9.   
  
  
  
  9.     CmdLine::addGlobalOptions( options , hidden );  
    10.   
  
  
  
  10.     //添加sharding选项描述信息  
    11.   
  
  
  
  11.     sharding_options.add_options()  
    12.   
  
  
  
  12.     ( "configdb" , po::value() , "1 or 3 comma separated config servers" )  
    13.   
  
  
  
  13.     ( "test" , "just run unit tests" )  
    14.   
  
  
  
  14.     ( "upgrade" , "upgrade meta data version" )  
    15.   
  
  
  
  15.     ( "chunkSize" , po::value(), "maximum amount of data per chunk" )  
    16.   
  
  
  
  16.     ( "ipv6", "enable IPv6 support (disabled by default)" )  
    17.   
  
  
  
  17.     ( "jsonp","allow JSONP access via http (has security implications)" )  
    18.   
  
  
  
  18.     ;  
    19.   
  
  
  
  19.     options.add(sharding_options);  
    20.   
  
  
  
  20.     ..... 
    21.

在完成option描述信息的初始化操作之后，下面就开始对启动命令行参数进行分析和执行了，如下：

 
 
 
 
      
  
  
  
  1. .....  
    2.   
  
  
  
  2.     // parse options  
    3.   
  
  
  
  3.     po::variables_map params;  
    4.   
  
  
  
  4.     //对argc，argv进行分析并转换成params，以便下面使用  
    5.   
  
  
  
  5.     if ( ! CmdLine::store( argc , argv , options , hidden , positional , params ) )  
    6.   
  
  
  
  6.         return 0;  
    7.   
  
  
  
  7.     // The default value may vary depending on compile options, but for mongos  
    8.   
  
  
  
  8.     // we want durability to be disabled.  
    9.   
  
  
  
  9.     cmdLine.dur = false;  
    10.   
  
  
  
  10.     //如果是help  
    11.   
  
  
  
  11.     if ( params.count( "help" ) ) {  
    12.   
  
  
  
  12.         cout << options << endl;  
    13.   
  
  
  
  13.         return 0;  
    14.   
  
  
  
  14.     }  
    15.   
  
  
  
  15.     //如果是版本信息  
    16.   
  
  
  
  16.     if ( params.count( "version" ) ) {  
    17.   
  
  
  
  17.         printShardingVersionInfo();  
    18.   
  
  
  
  18.         return 0;  
    19.   
  
  
  
  19.     }  
    20.   
  
  
  
  20.     //如要设置chunkSize  
    21.   
  
  
  
  21.     if ( params.count( "chunkSize" ) ) {  
    22.   
  
  
  
  22.         Chunk::MaxChunkSize = params["chunkSize"].as() * 1024 * 1024;  
    23.   
  
  
  
  23.     }  
    24.   
  
  
  
  24.     ......  
    25.   
  
  
  
  25.     //必选项，设置configdb信息  
    26.   
  
  
  
  26.     if ( ! params.count( "configdb" ) ) {  
    27.   
  
  
  
  27.        out() << "error: no args for --configdb" << endl;  
    28.   
  
  
  
  28.        return 4;  
    29.   
  
  
  
  29.     }  
    30.   
  
  
  
  30.     vector configdbs;  
    31.   
  
  
  
  31.     //对参数configdb进行分割 (以','分割 )  
    32.   
  
  
  
  32.     splitStringDelim( params["configdb"].as() , &configdbs , ',' );  
    33.   
  
  
  
  33.     //mongodb强制为1或3，具体原因不明  
    34.   
  
  
  
  34.     if ( configdbs.size() != 1 && configdbs.size() != 3 ) {  
    35.   
  
  
  
  35.         out() << "need either 1 or 3 configdbs" << endl;  
    36.   
  
  
  
  36.         return 5;  
    37.   
  
  
  
  37.     }  
    38.   
  
  
  
  38.     // we either have a seeting were all process are in localhost or none is  
    39.   
  
  
  
  39.     for ( vector::const_iterator it = configdbs.begin() ; it != configdbs.end() ; ++it ) {  
    40.   
  
  
  
  40.         try {  
    41.   
  
  
  
  41.             // 根据地址参数实例化HostAndPort对象，如地址不合法则抛出异常  
    42.   
  
  
  
  42.             HostAndPort configAddr( *it );  
    43.   
  
  
  
  43.             if ( it == configdbs.begin() ) {  
    44.   
  
  
  
  44.                 grid.setAllowLocalHost( configAddr.isLocalHost() );  
    45.   
  
  
  
  45.             }  
    46.   
  
  
  
  46.             //不允许在configdbs出现本地地址，注：如果configdb中全部为本地地址  
    47.   
  
  
  
  47.             //(实际用处不大)时不会执行下面if逻辑  
    48.   
  
  
  
  48.             if ( configAddr.isLocalHost() != grid.allowLocalHost() ) {  
    49.   
  
  
  
  49.                 out() << "cannot mix localhost and ip addresses in configdbs" << endl;  
    50.   
  
  
  
  50.                 return 10;  
    51.   
  
  
  
  51.             }  
    52.   
  
  
  
  52.         }  
    53.   
  
  
  
  53.         catch ( DBException& e) {  
    54.   
  
  
  
  54.             out() << "configdb: " << e.what() << endl;  
    55.   
  
  
  
  55.             return 9;  
    56.   
  
  
  
  56.         }  
    57.   
  
  
  
  57.     } 
    58.

上面完成了对命令行参数分析之后，接下来mongos要加载绑定几个hook：

 
 
 
 
      
  
  
  
  1. // set some global state  
    2.   
  
  
  
  2. //添加对链接池hook的绑定(shardingConnectionHook对象引用)，以最终调用其onHandedOut方法  
    3.   
  
  
  
  3. pool.addHook( &shardingConnectionHook );  
    4.   
  
  
  
  4. //设置链接池名称  
    5.   
  
  
  
  5. pool.setName( "mongos connectionpool" );  
    6.   
  
  
  
  6. //不设置“延迟kill游标”  
    7.   
  
  
  
  7. DBClientConnection::setLazyKillCursor( false );  
    8.   
  
  
  
  8. //设置当replicaSet配置修改时的hook对象(replicaSetChangey方法会更新链接对象信息  
    9.   
  
  
  
  9. ReplicaSetMonitor::setConfigChangeHook( boost::bind( &ConfigServer::replicaSetChange , &configServer , _1 ) ); 
    10.

上面的hook主要是在mongos主程序启动完成后，在运行期间执行一些数据操作时执行某些额外操作。从代码可以看出，mongos使用了链接池功能以提升获取链接的效率，具体实现机制我会在后绪章节中加以阐述。代码中的ReplicaSetMonitor类为一个维护和获取有效复制集的监视类，它提供了获取有效master,slave 的方法。完成这一步绑定后，接着mongos就会对config server信息进行初始化和升级操作了，如下:

 
 
 
 
      
  
  
  
  1. //显示sharding版本信息  
    2.   
  
  
  
  2. printShardingVersionInfo();  
    3.   
  
  
  
  3. //实始化configServer  
    4.   
  
  
  
  4. if ( ! configServer.init( configdbs ) ) {  
    5.   
  
  
  
  5.     cout << "couldn't resolve config db address" << endl;  
    6.   
  
  
  
  6.     return 7;  
    7.   
  
  
  
  7. }  
    8.   
  
  
  
  8. if ( ! configServer.ok( true ) ) {  
    9.   
  
  
  
  9.     cout << "configServer startup check failed" << endl;  
    10.   
  
  
  
  10.     return 8;  
    11.   
  
  
  
  11. }  
    12.   
  
  
  
  12. //检查Config版本信息(必要时进行升级操作)  
    13.   
  
  
  
  13. int configError = configServer.checkConfigVersion( params.count( "upgrade" ) );  
    14.   
  
  
  
  14. if ( configError ) {  
    15.   
  
  
  
  15.     if ( configError > 0 ) {  
    16.   
  
  
  
  16.         cout << "upgrade success!" << endl;  
    17.   
  
  
  
  17.     }  
    18.   
  
  
  
  18.     else {  
    19.   
  
  
  
  19.         cout << "config server error: " << configError << endl;  
    20.   
  
  
  
  20.     }  
    21.   
  
  
  
  21.     return configError;  
    22.   
  
  
  
  22. }  
    23.   
  
  
  
  23. //重新设置config db信息(包括shard中chunk的min，lastmod信息)  
    24.   
  
  
  
  24. configServer.reloadSettings(); 
    25.

***就是启动侦听服务，这里mongos启动了两个侦听服务器，一个是以线程方式启动，用于接收授权的用户操作信息，另一个则是普遍的循环侦听服务，用于侦听客户端message如下：

 
 
 
 
      
  
  
  
  1. //初始化一些Signals信息，用于处理程序退出，中断等情况  
    2.   
  
  
  
  2. init();  
    3.   
  
  
  
  3. //以线程方式启动webserver，循环侦听授权访问的 message信息，详见dbwebserver.cpp文件中allowed方法  
    4.   
  
  
  
  4. boost::thread web( boost::bind(&webServerThread, new NoAdminAccess() /* takes ownership */) );  
    5.   
  
  
  
  5. MessageServer::Options opts;  
    6.   
  
  
  
  6. opts.port = cmdLine.port;  
    7.   
  
  
  
  7. opts.ipList = cmdLine.bind_ip;  
    8.   
  
  
  
  8. start(opts);//启动message服务器，侦听客户端message  
    9.   
  
  
  
  9. dbexit( EXIT_CLEAN );  
    10.   
  
  
  
  10. return 0; 
    11.

到这里，main代码就介绍完了，但上面代码段中的start才是启动balancer来均衡各个shard间chunk的操作，所以我们接着再看一下该方法的实现：

 
 
 
 
      
  
  
  
  1. void start( const MessageServer::Options& opts ) {  
    2.   
  
  
  
  2.     setThreadName( "mongosMain" );//设置线程名称  
    3.   
  
  
  
  3.     installChunkShardVersioning();//绑定chunk shard版本控制信息  
    4.   
  
  
  
  4.     balancer.go();//均衡shard 中chunk(节点)信息，详情参见 balance.cpp的run()方法  
    5.   
  
  
  
  5.     cursorCache.startTimeoutThread();//对空闲(过期)游标进行清除操作  
    6.   
  
  
  
  6.     log() << "waiting for connections on port " << cmdLine.port << endl;  
    7.   
  
  
  
  7.     ShardedMessageHandler handler;  
    8.   
  
  
  
  8.     MessageServer * server = createServer( opts , &handler );//构造server对象  
    9.   
  
  
  
  9.     server->setAsTimeTracker();  
    10.   
  
  
  
  10.     server->run();//启动message服务  
    11.   
  
  
  
  11. } 
    12.

好了，今天的内容到这里就告一段落了，在接下来的文章中，将会介绍balancer的实现方式和操作流程。

原文链接:http://www.cnblogs.com/daizhj/archive/2011/05/16/2022041.html

【编辑推荐】

  1. Mongodb源码分析--内存文件映射(MMAP)
  2. 走进MongoDB的世界 展开MongoDB的学习之旅
  3. 浅析Mongodb源码之游标Cursor
  4. 野心勃勃的NoSQL新贵 MongoDB应用实战
  5. MongoDB与CouchDB全方位对比

分享题目：Mongodb源码分析之Mongos分析
链接URL：http://www.shufengxianlan.com/qtweb/news31/132181.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

自适应网站知识

各行业网站

app网站建设　　 　软文推广公司　　 　宏昇商务服务　　 　云主机　　 　成都网站设计　　 　手机网站建设套餐　　 　橡塑板　　 　德阳服务器托管　　 　成都网站建设　　 　安防网站建设方案　　 　公路打桩机　　 　郫县明安消防器材　　 　食品包装设计　　 　成都花箱　　 　成都水电改造　　 　和县翔豪网站　　 　企业网站改版　　 　成都托管服务器　　 　仿古建筑设计　　 　资阳网站建设