MongoDB集群和实战详解

1.概述

任县网站制作公司哪家好，找成都创新互联！从网页设计、网站建设、微信开发、APP开发、成都响应式网站建设公司等网站项目制作，到程序开发，运营维护。成都创新互联于2013年开始到现在10年的时间，我们拥有了丰富的建站经验和运维经验，来保证我们的工作的顺利进行。专注于网站建设就选成都创新互联。

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

  • 基本操作
  • CRUD
  • MapReduce

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

2.基本操作

常用的 Shell 命令如下所示：

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

查看集合基本信息，内容如下所示：

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

3.CRUD

3.1创建

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

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

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

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

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(); 
    2.

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

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

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

3.3.3查询等于条件数据

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

3.3.4查询大于条件数据

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

3.3.5查询小于条件数据

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

3.3.6查询大于等于条件数据

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

3.3.7查询小于等于条件数据

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

3.3.8查询AND和OR条件数据

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

3.3.9模糊查询

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

3.3.10开头匹配

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

3.3.11指定列查询

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

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

3.3.12指定列查询+条件查询

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

3.3.13排序

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

3.3.14查询5条数据

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

3.3.15N条以后数据

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

3.3.16在一定区域内查询记录

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

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

3.3.17COUNT

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

3.3.18安装结果集排序

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

3.3.19不等于NULL

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

3.4索引

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

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

3.5更新

update命令格式，如下所示：

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

参数说明： criteria：

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

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

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

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

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

3.6删除

3.6.1删除指定记录

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

3.6.2删除所有记录

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

3.6.3DROP

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

4.MapReduce

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

MongdbManager类，用来初始化MongoDB：

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

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

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

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

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

5.总结

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

网页标题：MongoDB集群和实战详解
本文网址：http://www.shufengxianlan.com/qtweb/news42/251192.html

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

广告

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

猜你还喜欢下面的内容

网站制作知识

行业网站建设

众诚伟业　　 　成都网站设计　　 　网站制作公司　　 　宣威网站建设　　 　成都网站营销　　 　雕琢时光茶叶　　 　成都户外广告牌设计　　 　葆金莱自动门　　 　背发光字　　 　营销型网站建设　　 　电信成都枢纽中心　　 　眉山电信机房　　 　腾讯云香港空间　　 　成都app软件开发　　 　企业画册设计制作　　 　成都网站建设公司　　 　成都微信开发　　 　vi形象设计　　 　成都网站设计　　 　定制网站制作