MongoDB集群和实战详解

本文涉及的产品
云数据库 MongoDB,通用型 2核4GB
云数据库 MongoDB,独享型 2核8GB
推荐场景:
构建全方位客户视图
简介:

1.概述

最近有同学和网友私信我,问我MongoDB方面的问题;这里我整理一篇博客来赘述下MongoDB供大家学习参考,博客的目录内容如下:

  • 基本操作
  • CRUD
  • MapReduce

本篇文章是基于MongoDB集群(Sharding+Replica Sets)上演示的,故操作的内容都是集群层面的,所以有些命令和单独的使用MongoDB库有异样。

2.基本操作

常用的 Shell 命令如下所示:

 
  1. db.help()    # 数据库帮助 
  2. db.collections.help()    # 集合帮助 
  3. rs.help()    # help on replica set 
  4. show dbs    # 展示数据库名 
  5. show collections    # 展示collections在当前库 
  6. use db_name    # 选择数据库 

查看集合基本信息,内容如下所示:

 
  1. #查看帮助  
  2. db.yourColl.help(); 
  3.  
  4. #查询当前集合的数据条数  
  5. db.yourColl.count(); 
  6.  
  7. #查看数据空间大小  
  8. db.userInfo.dataSize(); 
  9.  
  10. #得到当前聚集集合所在的 
  11. db db.userInfo.getDB(); 
  12.  
  13. #得到当前聚集的状态  
  14. db.userInfo.stats(); 
  15.  
  16. #得到聚集集合总大小  
  17. db.userInfo.totalSize(); 
  18.  
  19. #聚集集合储存空间大小  
  20. db.userInfo.storageSize(); 
  21.  
  22. #Shard版本信息  
  23. db.userInfo.getShardVersion() 
  24.  
  25. #聚集集合重命名,将userInfo重命名为users 
  26. db.userInfo.renameCollection("users");  
  27.   
  28. #删除当前聚集集合  
  29. db.userInfo.drop(); 

3.CRUD

3.1创建

在集群中,我们增加一个 friends 库,命令如下所示:

 
  1. db.runCommand({enablesharding:"friends"}); 

在库新建后,我们在该库下创建一个user分片,命令如下:

 
  1. db.runCommand( { shardcollection : "friends. user"}); 

3.2新增

在MongoDB中,save和insert都能达到新增的效果。但是这两者是有区别的,在save函数中,如果原来的对象不存在,那他们都可以向collection里插入数据;如果已经存在,save会调用update更新里面的记录,而insert则会忽略操作。

另外,在insert中可以一次性插叙一个列表,而不用遍历,效率高,save则需要遍历列表,一个个插入,下面我们可以看下两个函数的原型,通过函数原型我们可以看出,对于远程调用来说,是一次性将整个列表post过来让MongoDB去处理,效率会高些。

Save函数原型如下所示:

Insert函数原型(部分代码)如下所示:

3.3查询

3.3.1查询所有记录

 
  1. db. user.find(); 

默认每页显示20条记录,当显示不下的情况下,可以用it迭代命令查询下一页数据。注意:键入it命令不能带“;” 但是你可以设置每页显示数据的大小,用DBQuery.shellBatchSize= 50;这样每页就显示50条记录了。

3.3.2查询去掉后的当前聚集集合中的某列的重复数据

 
  1. db. user.distinct("name");  
  2.  
  3. #会过滤掉name中的相同数据 相当于: 
  4. select distict name from user

3.3.3查询等于条件数据

 
  1. db.user.find({"age": 24});  
  2. #相当于:  
  3. select * from user where age = 24; 

3.3.4查询大于条件数据

 
  1. db.user.find({age: {$gt: 24}});  
  2.  
  3. # 相当于: 
  4. select * from user where age >24; 

3.3.5查询小于条件数据

 
  1. db.user.find({age: {$lt: 24}});  
  2. #相当于: 
  3. select * from user where age < 24; 

3.3.6查询大于等于条件数据

 
  1. db.user.find({age: {$gte: 24}});  
  2. #相当于: 
  3. select * from user where age >= 24; 

3.3.7查询小于等于条件数据

 
  1. db.user.find({age: {$lte: 24}});  
  2. #相当于: 
  3. select * from user where age <= 24; 

3.3.8查询AND和OR条件数据

  • AND
 
  1. db.user.find({age: {$gte: 23, $lte: 26}}); 
  2.  
  3. #相当于 
  4. select * from user where age >=23 and age <= 26; 
  • OR
 
  1. db.user.find({$or: [{age: 22}, {age: 25}]});  
  2.  
  3. #相当于: 
  4. select * from user where age = 22 or age = 25; 

3.3.9模糊查询

 
  1. db.user.find({name: /mongo/});  
  2.  
  3. #相当于%%  
  4. select * from user where name like '%mongo%'

3.3.10开头匹配

 
  1. db.user.find({name: /^mongo/});  
  2. # 与SQL中得like语法类似 
  3. select * from user where name like 'mongo%'

3.3.11指定列查询

 
  1. db.user.find({}, {name: 1, age: 1});  
  2.  
  3. #相当于: 
  4. select name, age from user

当然name也可以用true或false,当用ture的情况下和name:1效果一样,如果用false就是排除name,显示name以外的列信息。

3.3.12指定列查询+条件查询

 
  1. db.user.find({age: {$gt: 25}}, {name: 1, age: 1});  
  2.  
  3. #相当于: 
  4. select name, age from user where age > 25; 
  5.  
  6.  db.user.find({name'zhangsan', age: 22}); 
  7.  
  8.  #相当于: 
  9.  
  10.  select * from user where name = 'zhangsan' and age = 22; 

3.3.13排序

 
  1. #升序: 
  2. db.user.find().sort({age: 1});  
  3.  
  4. #降序: 
  5. db.user.find().sort({age: -1}); 

3.3.14查询5条数据

 
  1. db.user.find().limit(5);  
  2.  
  3. #相当于: 
  4. select * from user limit 5; 

3.3.15N条以后数据

 
  1. db.user.find().skip(10);  
  2.  
  3. #相当于: 
  4. select * from user where id not in ( select * from user limit 5 ); 

3.3.16在一定区域内查询记录

 
  1. #查询在5~10之间的数据 
  2. db.user.find().limit(10).skip(5); 

可用于分页,limit是pageSize,skip是第几页*pageSize。

3.3.17COUNT

 
  1. db.user.find({age: {$gte: 25}}).count();  
  2.  
  3. #相当于: 
  4. select count(*) from user where age >= 20; 

3.3.18安装结果集排序

 
  1. db.userInfo.find({sex: {$exists: true}}).sort(); 

3.3.19不等于NULL

 
  1. db.user.find({sex: {$ne: null}})  
  2.  
  3. #相当于: 
  4. select * from user where sex not null

3.4索引

创建索引,并指定主键字段,命令内容如下所示:

 
  1. db.epd_favorites_folder.ensureIndex({"id":1},{"unique":true,"dropDups":true})  
  2. db.epd_focus.ensureIndex({"id":1},{"unique":true,"dropDups":true}) 

3.5更新

update命令格式,如下所示:

 
  1. db.collection.update(criteria,objNew,upsert,multi) 

参数说明: criteria:

查询条件 objNew:update对象和一些更新操作符

upsert:如果不存在update的记录,是否插入objNew这个新的文档,true为插入,默认为false,不插入。

multi:默认是false,只更新找到的第一条记录。如果为true,把按条件查询出来的记录全部更新。

下面给出一个示例,更新id为 1 中 price 的值,内容如下所示:

 
  1. db. user.update({id: 1},{$set:{price:2}});   
  2.  
  3. #相当于: 
  4. update user set price=2 where id=1; 

3.6删除

3.6.1删除指定记录

 
  1. db. user. remove( { id:1 } );  
  2.  
  3. #相当于:  
  4. delete from user where id=1; 

3.6.2删除所有记录

 
  1. db. user. remove( { } );   
  2.  
  3. #相当于: 
  4. delete from user

3.6.3DROP

 
  1. db. userdrop();   
  2.  
  3. #相当于: 
  4. drop table user

4.MapReduce

MongoDB中的 MapReduce 是编写JavaScript脚本,然后由MongoDB去解析执行对应的脚本,下面给出 Java API 操作MR。代码如下所示:

MongdbManager类,用来初始化MongoDB:

 
  1. package cn.mongo.util; 
  2.  
  3. import org.slf4j.Logger; 
  4. import org.slf4j.LoggerFactory; 
  5.  
  6. import com.mongodb.DB; 
  7. import com.mongodb.Mongo; 
  8. import com.mongodb.MongoOptions; 
  9.  
  10. /** 
  11.  * @Date Mar 3, 2015 
  12.  *  
  13.  * @author dengjie 
  14.  *  
  15.  * @Note mongodb manager 
  16.  */ 
  17. public class MongdbManager { 
  18.  
  19.     private static final Logger logger = LoggerFactory.getLogger(MongdbManager.class); 
  20.     private static Mongo mongo = null
  21.     private static String tag = SystemConfig.getProperty("dev.tag"); 
  22.  
  23.     private MongdbManager() { 
  24.     } 
  25.  
  26.     static { 
  27.         initClient(); 
  28.     } 
  29.  
  30.     // get DB object 
  31.     public static DB getDB(String dbName) { 
  32.         return mongo.getDB(dbName); 
  33.     } 
  34.  
  35.     // get DB object without param 
  36.     public static DB getDB() { 
  37.         String dbName = SystemConfig.getProperty(String.format("%s.mongodb.dbname", tag)); 
  38.         return mongo.getDB(dbName); 
  39.     } 
  40.  
  41.     // init mongodb pool 
  42.     private static void initClient() { 
  43.         try { 
  44.             String[] hosts = SystemConfig.getProperty(String.format("%s.mongodb.host", tag)).split(","); 
  45.             for (int i = 0; i < hosts.length; i++) { 
  46.                 try { 
  47.                     String host = hosts[i].split(":")[0]; 
  48.                     int port = Integer.parseInt(hosts[i].split(":")[1]); 
  49.                     mongo = new Mongo(host, port); 
  50.                     if (mongo.getDatabaseNames().size() > 0) { 
  51.                         logger.info(String.format("connection success,host=[%s],port=[%d]", host, port)); 
  52.                         break; 
  53.                     } 
  54.                 } catch (Exception ex) { 
  55.                     ex.printStackTrace(); 
  56.                     logger.error(String.format("create connection has error,msg is %s", ex.getMessage())); 
  57.                 } 
  58.             } 
  59.  
  60.             // 设置连接池的信息 
  61.             MongoOptions opt = mongo.getMongoOptions(); 
  62.             opt.connectionsPerHost = SystemConfig.getIntProperty(String.format("%s.mongodb.poolsize", tag));// poolsize 
  63.             opt.threadsAllowedToBlockForConnectionMultiplier = SystemConfig.getIntProperty(String.format( 
  64.                     "%s.mongodb.blocksize", tag));// blocksize 
  65.             opt.socketKeepAlive = true
  66.             opt.autoConnectRetry = true
  67.         } catch (Exception e) { 
  68.             e.printStackTrace(); 
  69.         } 
  70.     } 

MongoDBFactory类,用来封装操作业务代码,具体内容如下所示:

 
  1. package cn.mongo.util; 
  2.  
  3. import java.util.ArrayList; 
  4. import java.util.List; 
  5.  
  6. import org.slf4j.Logger; 
  7. import org.slf4j.LoggerFactory; 
  8.  
  9. import cn.diexun.domain.MGDCustomerSchema; 
  10.  
  11. import com.mongodb.BasicDBList; 
  12. import com.mongodb.DB; 
  13. import com.mongodb.DBCollection; 
  14. import com.mongodb.DBObject; 
  15. import com.mongodb.util.JSON; 
  16.  
  17. /** 
  18.  * @Date Mar 3, 2015 
  19.  * 
  20.  * @Author dengjie 
  21.  */ 
  22. public class MongoDBFactory { 
  23.  
  24.     private static Logger logger = LoggerFactory.getLogger(MongoDBFactory.class); 
  25.  
  26.     // save data to mongodb 
  27.     public static void save(MGDCustomerSchema mgs, String collName) { 
  28.         DB db = null
  29.         try { 
  30.             db = MongdbManager.getDB(); 
  31.             DBCollection coll = db.getCollection(collName); 
  32.             DBObject dbo = (DBObject) JSON.parse(mgs.toString()); 
  33.             coll.insert(dbo); 
  34.         } catch (Exception ex) { 
  35.             ex.printStackTrace(); 
  36.             logger.error(String.format("save object to mongodb has error,msg is %s", ex.getMessage())); 
  37.         } finally { 
  38.             if (db != null) { 
  39.                 db.requestDone(); 
  40.                 db = null
  41.             } 
  42.         } 
  43.     } 
  44.  
  45.     // batch insert 
  46.     public static void save(List<?> mgsList, String collName) { 
  47.         DB db = null
  48.         try { 
  49.             db = MongdbManager.getDB(); 
  50.             DBCollection coll = db.getCollection(collName); 
  51.             BasicDBList data = (BasicDBList) JSON.parse(mgsList.toString()); 
  52.             List<DBObject> list = new ArrayList<DBObject>(); 
  53.             int commitSize = SystemConfig.getIntProperty("mongo.commit.size"); 
  54.             int rowCount = 0; 
  55.             long start = System.currentTimeMillis(); 
  56.             for (Object dbo : data) { 
  57.                 rowCount++; 
  58.                 list.add((DBObject) dbo); 
  59.                 if (rowCount % commitSize == 0) { 
  60.                     try { 
  61.                         coll.insert(list); 
  62.                         list.clear(); 
  63.                         logger.info(String.format("current commit rowCount = [%d],commit spent time = [%s]s", rowCount, 
  64.                                 (System.currentTimeMillis() - start) / 1000.0)); 
  65.                     } catch (Exception ex) { 
  66.                         ex.printStackTrace(); 
  67.                         logger.error(String.format("batch commit data to mongodb has error,msg is %s", ex.getMessage())); 
  68.                     } 
  69.                 } 
  70.             } 
  71.             if (rowCount % commitSize != 0) { 
  72.                 try { 
  73.                     coll.insert(list); 
  74.                     logger.info(String.format("insert data to mongo has spent total time = [%s]s"
  75.                             (System.currentTimeMillis() - start) / 1000.0)); 
  76.                 } catch (Exception ex) { 
  77.                     ex.printStackTrace(); 
  78.                     logger.error(String.format("commit end has error,msg is %s", ex.getMessage())); 
  79.                 } 
  80.             } 
  81.         } catch (Exception ex) { 
  82.             ex.printStackTrace(); 
  83.             logger.error(String.format("save object list to mongodb has error,msg is %s", ex.getMessage())); 
  84.         } finally { 
  85.             if (db != null) { 
  86.                 db.requestDone(); 
  87.                 db = null
  88.             } 
  89.         } 
  90.     } 

LoginerAmountMR类,这是一个统计登录用户数的MapReduce计算类,代码如下:

 
  1. package cn.mongo.mapreduce; 
  2.  
  3. import java.sql.Timestamp
  4. import java.util.ArrayList; 
  5. import java.util.Date
  6. import java.util.List; 
  7.  
  8. import org.bson.BSONObject; 
  9. import org.slf4j.Logger; 
  10. import org.slf4j.LoggerFactory; 
  11.  
  12. import cn.diexun.conf.ConfigureAPI.MR; 
  13. import cn.diexun.conf.ConfigureAPI.PRECISION
  14. import cn.diexun.domain.Kpi; 
  15. import cn.diexun.util.CalendarUtil; 
  16. import cn.diexun.util.MongdbManager; 
  17. import cn.diexun.util.MysqlFactory; 
  18.  
  19. import com.mongodb.DB; 
  20. import com.mongodb.DBCollection; 
  21. import com.mongodb.DBCursor; 
  22. import com.mongodb.DBObject; 
  23. import com.mongodb.MapReduceOutput; 
  24. import com.mongodb.ReadPreference; 
  25.  
  26. /** 
  27.  * @Date Mar 13, 2015 
  28.  *  
  29.  * @Author dengjie 
  30.  *  
  31.  * @Note use mr jobs stats user login amount 
  32.  */ 
  33. public class LoginerAmountMR { 
  34.     private static Logger logger = LoggerFactory.getLogger(LoginerAmountMR.class); 
  35.  
  36.  // map 函数JS字符串拼接  
  37.     private static String map() { 
  38.         String map = "function(){"
  39.         map += "if(this.userName != \"\"){"
  40.         map += "emit({" + "kpi_code:'login_times',username:this.userName," 
  41.                 + "district_id:this.districtId,product_style:this.product_style," 
  42.                 + "customer_property:this.customer_property},{count:1});"
  43.         map += "}"
  44.         map += "}"
  45.         return map; 
  46.     } 
  47.  
  48.  
  49.     private static String reduce() { 
  50.         String reduce = "function(key,values){"
  51.         reduce += "var total = 0;"
  52.         reduce += "for(var i=0;i<values.length;i++){"
  53.         reduce += "total += values[i].count;}"
  54.         reduce += "return {count:total};"
  55.         reduce += "}"
  56.         return reduce; 
  57.     } 
  58.  
  59. // reduce 函数字符串拼接 
  60.     public static void main(String[] args) { 
  61.         loginNumbers("t_login_20150312"); 
  62.     } 
  63.  
  64.     /** 
  65.      * login user amount 
  66.      *  
  67.      * @param collName 
  68.      */ 
  69.     public static void loginNumbers(String collName) { 
  70.         DB db = null
  71.         try { 
  72.             db = MongdbManager.getDB(); 
  73.             db.setReadPreference(ReadPreference.secondaryPreferred()); 
  74.             DBCollection coll = db.getCollection(collName); 
  75.             String result = MR.COLLNAME_TMP; 
  76.  
  77.             long start = System.currentTimeMillis(); 
  78.             MapReduceOutput mapRed = coll.mapReduce(map(), reduce(), result, null); 
  79.             logger.info(String.format("mr run spent time=%ss", (System.currentTimeMillis() - start) / 1000.0)); 
  80.             start = System.currentTimeMillis(); 
  81.             DBCursor cursor = mapRed.getOutputCollection().find(); 
  82.             List<Kpi> list = new ArrayList<Kpi>(); 
  83.             while (cursor.hasNext()) { 
  84.                 DBObject obj = cursor.next(); 
  85.                 BSONObject key = (BSONObject) obj.get("_id"); 
  86.                 BSONObject value = (BSONObject) obj.get("value"); 
  87.                 Object kpiValue = value.get("count"); 
  88.                 Object userName = key.get("username"); 
  89.                 Object districtId = key.get("district_id"); 
  90.                 Object customerProperty = key.get("customer_property"); 
  91.                 Object productStyle = key.get("product_style"); 
  92.                 Kpi kpi = new Kpi(); 
  93.                 try { 
  94.                     kpi.setUserName(userName == null ? "" : userName.toString()); 
  95.                     kpi.setKpiCode(key.get("kpi_code").toString()); 
  96.                     kpi.setKpiValue(Math.round(Double.parseDouble(kpiValue.toString()))); 
  97.                     kpi.setCustomerProperty(customerProperty == null ? "" : customerProperty.toString()); 
  98.                     kpi.setDistrictId(districtId == "" ? 0 : Integer.parseInt(districtId.toString())); 
  99.                     kpi.setProductStyle(productStyle == null ? "" : productStyle.toString()); 
  100.                     kpi.setCreateDate(collName.split("_")[2]); 
  101.                     kpi.setUpdateDate(Timestamp.valueOf(CalendarUtil.formatMap.get(PRECISION.HOUR).format(new Date()))); 
  102.                     list.add(kpi); 
  103.                 } catch (Exception exx) { 
  104.                     exx.printStackTrace(); 
  105.                     logger.error(String.format("parse type or get value has error,msg is %s", exx.getMessage())); 
  106.                 } 
  107.             } 
  108.             MysqlFactory.insert(list); 
  109.             logger.info(String.format("store mysql spent time is %ss", (System.currentTimeMillis() - start) / 1000.0)); 
  110.         } catch (Exception ex) { 
  111.             ex.printStackTrace(); 
  112.             logger.error(String.format("run map-reduce jobs has error,msg is %s", ex.getMessage())); 
  113.         } finally { 
  114.             if (db != null) { 
  115.                 db.requestDone(); 
  116.                 db = null
  117.             } 
  118.         } 
  119.     } 

5.总结

在计算 MongoDB 的MapReduce计算的时候,拼接JavaScript字符串时需要谨慎小心,很容易出错,上面给出的代码只是一部分代码,供参考学习使用;另外,若是要做MapReduce任务计算,推荐使用Hadoop的MapReduce计算框架,MongoDB的MapReduce框架这里仅做介绍学习了解。


本文作者:佚名

来源:51CTO

相关实践学习
MongoDB数据库入门
MongoDB数据库入门实验。
快速掌握 MongoDB 数据库
本课程主要讲解MongoDB数据库的基本知识,包括MongoDB数据库的安装、配置、服务的启动、数据的CRUD操作函数使用、MongoDB索引的使用(唯一索引、地理索引、过期索引、全文索引等)、MapReduce操作实现、用户管理、Java对MongoDB的操作支持(基于2.x驱动与3.x驱动的完全讲解)。 通过学习此课程,读者将具备MongoDB数据库的开发能力,并且能够使用MongoDB进行项目开发。 &nbsp; 相关的阿里云产品:云数据库 MongoDB版 云数据库MongoDB版支持ReplicaSet和Sharding两种部署架构,具备安全审计,时间点备份等多项企业能力。在互联网、物联网、游戏、金融等领域被广泛采用。 云数据库MongoDB版(ApsaraDB for MongoDB)完全兼容MongoDB协议,基于飞天分布式系统和高可靠存储引擎,提供多节点高可用架构、弹性扩容、容灾、备份回滚、性能优化等解决方案。 产品详情: https://www.aliyun.com/product/mongodb
相关文章
|
2月前
|
NoSQL Java 测试技术
|
2月前
|
NoSQL Java 测试技术
spring boot MongoDB实战(二)
spring boot MongoDB实战
138 1
|
2月前
|
NoSQL Java MongoDB
Spring Boot中MongoDB的使用和实战
Spring Boot中MongoDB的使用和实战
125 0
|
10天前
|
自然语言处理 运维 NoSQL
MongoDB集群同步
实现 MongoDB Cluster-to-Cluster 即集群同步的工具是:mongosync 详情可参考如下官方文档: https://www.mongodb.com/zh-cn/docs/cluster-to-cluster-sync/current/quickstart/ 以上这个地址的文档一看就是机器翻译的,可能有不恰当的地方,但基本可参考使用。 以下是本次在某项目地配置集群同步的简要步骤,可参考使用。
24 6
|
26天前
|
存储 NoSQL MongoDB
MongoDB实战面试指南:常见问题一网打尽
MongoDB实战面试指南:常见问题一网打尽
|
1月前
|
存储 负载均衡 NoSQL
MongoDB的架构设计基于三种集群模式
【6月更文挑战第5天】MongoDB的架构设计基于三种集群模式
36 3
|
2月前
|
存储 NoSQL 关系型数据库
MongoDB非关系型数据库实战
【5月更文挑战第6天】MongoDB,流行的NoSQL数据库,以其灵活的数据模型和高性能备受青睐。本文介绍了MongoDB的基础,包括文档型数据库特性、安装配置、数据操作。通过电商订单管理的实战案例,展示了MongoDB在处理复杂数据结构和大规模数据时的优势,适用于电商、游戏、视频直播等场景。MongoDB的索引、全文搜索和地理空间功能进一步增强了其实用性。注意性能优化和扩展性以确保系统稳定性和可靠性。
|
2月前
|
NoSQL Java MongoDB
Spring Boot 整合 MongoDB 实战
本文介绍了如何使用Spring Boot整合MongoDB,实现数据持久化。步骤包括:环境准备(安装Java、MongoDB及创建Spring Boot项目)、在pom.xml中添加MongoDB依赖、配置MongoDB连接信息、创建映射MongoDB集合的实体类、定义MongoDB Repository接口、编写业务逻辑和服务层以及控制器层。通过测试确保整合成功。此实战教程帮助读者理解Spring Boot与MongoDB整合的基础,适用于快速构建Java应用。
473 11
|
2月前
|
存储 缓存 NoSQL
MongoDB详解(三)——MongoDB集群
MongoDB详解(三)——MongoDB集群
51 4
|
2月前
|
NoSQL MongoDB
搭建MongoDB分片式集群
搭建MongoDB分片式集群
48 0