基本介绍

什么是NoSQL数据库

NoSQL，指的是非关系型的数据库。NoSQL有时也称作Not Only SQL的缩写，是对不同于传统的关系型数据库的数据库管理系统的统称。

NoSQL用于超大规模数据的存储。（例如谷歌或Facebook每天为他们的用户收集万亿比特的数据）。这些类型的数据存储不需要固定的模式，无需多余操作就可以横向扩展。

为什么使用NoSQL数据库

今天我们可以通过第三方平台（如：Google,Facebook等）可以很容易的访问和抓取数据。用户的个人信息，社交网络，地理位置，用户生成的数据和用户操作日志已经成倍的增加。我们如果要对这些用户数据进行挖掘，那SQL数据库已经不适合这些应用了, NoSQL 数据库的发展却能很好的处理这些大的数据。

什么是MongoDB

在计算机科学中，CAP定理（CAP theorem），它指出对于一个分布式计算系统来说，不可能同时满足以下三点：

• 一致性(Consistency) (所有节点在同一时间具有相同的数据)
• 可用性(Availability) (保证每个请求不管成功或者失败都有响应)
• 分隔容忍(Partition tolerance) (系统中任意信息的丢失或失败不会影响系统的继续运作)

image.png

MongoDB 是由C++语言编写的，是一个基于分布式文件存储的开源数据库系统。
在高负载的情况下，添加更多的节点，可以保证服务器性能。
MongoDB 旨在为WEB应用提供可扩展的高性能数据存储解决方案。
MongoDB 将数据存储为一个文档，数据结构由键值(key=>value)对组成。MongoDB 文档类似于 JSON 对象。字段值可以包含其他文档，数组及文档数组。

MongoDB 相比 RDBMS 的优势

• 模式较少：MongoDB 是一种文档数据库，一个集合可以包含各种不同的文档。每个文档的字段数、内容以及文档大小都可以各不相同。
• 采用单个对象的模式，清晰简洁。
• 没有复杂的连接功能。
• 深度查询功能。MongoDB 支持对文档执行动态查询，使用的是一种不逊色于 SQL 语言的基于文档的查询语言。
• 具有调优功能。
• 易于扩展。MongoDB 非常易于扩展。
• 不需要从应用对象到数据库对象的转换/映射。
• 使用内部存储存储（窗口化）工作集，能够更快地访问数据。

为何选择使用 MongoDB

• 面向文档的存储：以 JSON 格式的文档保存数据。
• 任何属性都可以建立索引。
• 复制以及高可扩展性。
• 自动分片。
• 丰富的查询功能。
• 快速的即时更新。
• 来自 MongoDB 的专业支持。

MongoDB 适用的领域

• 大数据
• 内容管理及交付
• 移动及社会化基础设施
• 用户数据管理
• 数据中心

MongoDB适用场景

1）网站实时数据处理。它非常适合实时的插入、更新与查询，并具备网站实时数据存储所需的复制及高度伸缩性。
2）缓存。由于性能很高，它适合作为信息基础设施的缓存层。在系统重启之后，由它搭建的持久化缓存层可以避免下层的数据源过载。
3）高伸缩性的场景。非常适合由数十或数百台服务器组成的数据库，它的路线图中已经包含对MapReduce引擎的内置支持。

MongoDB不适用的场景

1）要求高度事务性的系统。
2）传统的商业智能应用。
3）复杂的跨文档（表）级联查询。

主要特点

• MongoDB 是一个面向文档存储的数据库，操作起来比较简单和容易。
• 你可以在MongoDB记录中设置任何属性的索引 (如：FirstName="Sameer",Address="8 Gandhi Road")来实现更快的排序。
• 你可以通过本地或者网络创建数据镜像，这使得MongoDB有更强的扩展性。
• 如果负载的增加（需要更多的存储空间和更强的处理能力） ，它可以分布在计算机网络中的其他节点上这就是所谓的分片。
• Mongo支持丰富的查询表达式。查询指令使用JSON形式的标记，可轻易查询文档中内嵌的对象及数组。
• MongoDb 使用update()命令可以实现替换完成的文档（数据）或者一些指定的数据字段 。
• Mongodb中的Map/reduce主要是用来对数据进行批量处理和聚合操作。
• Map和Reduce。Map函数调用emit(key,value)遍历集合中所有的记录，将key与value传给Reduce函数进行处理。
• Map函数和Reduce函数是使用Javascript编写的，并可以通过db.runCommand或mapreduce命令来执行MapReduce操作。
• GridFS是MongoDB中的一个内置功能，可以用于存放大量小文件。
• MongoDB允许在服务端执行脚本，可以用Javascript编写某个函数，直接在服务端执行，也可以把函数的定义存储在服务端，下次直接调用即可。
• MongoDB支持各种编程语言:RUBY，PYTHON，JAVA，C++，PHP，C#等多种语言。
• MongoDB安装简单。

数据模型

一个MongoDB 实例可以包含一组DataBase（数据库）；
一个DataBase 可以包含一组Collection（集合）；
一个Collection可以包含一组Document（文档）；
一个Document包含一组field（字段）；
一个字段是一个key/value pair。

key：必须为字符串类型，
value：可以包含如下内容：（string，int，float，timestamp，binary，document，数组）

通过下图实例，我们也可以更直观的了解Mongo中的一些概念：

image.png

文档范例：

{
   _id: ObjectId(7df78ad8902c)
   title: 'MongoDB Overview', 
   description: 'MongoDB is no sql database',
   by: 'tutorials point',
   url: 'http://www.tutorialspoint.com',
   tags: ['mongodb', 'database', 'NoSQL'],
   likes: 100, 
   comments: [  
      {
         user:'user1',
         message: 'My first comment',
         dateCreated: new Date(2011,1,20,2,15),
         like: 0 
      },
      {
         user:'user2',
         message: 'My second comments',
         dateCreated: new Date(2011,1,25,7,45),
         like: 5
      }
   ]
} 

基本安装

MongoDB安装

https://www.mongodb.com/download-center/community

image.png

可视化工具安装

NoSQLBooster for MongoDB https://nosqlbooster.com/downloads

image.png

image.png

• admin： 从权限的角度来看，这是"root"数据库。要是将一个用户添加到这个数据库，这个用户自动继承所有数据库的权限。一些特定的服务器端命令也只能从这个数据库运行，比如列出所有的数据库或者关闭服务器。
• local: 这个数据永远不会被复制，可以用来存储限于本地单台服务器的任意集合
• config: 当Mongo用于分片设置时，config数据库在内部使用，用于保存分片的相关信息。

MongoDB支持的数据类型

ObjectId
ObjectId 类似唯一主键，可以很快的去生成和排序，包含 12 bytes，含义是：

• 前 4 个字节表示创建 unix 时间戳,格林尼治时间 UTC 时间，比北京时间晚了 8 个小时
• 接下来的 3 个字节是机器标识码
• 紧接的两个字节由进程 id 组成 PID

• 最后三个字节是随机数

  
  
  image.png
  
  

  MongoDB 中存储的文档必须有一个 _id 键。这个键的值可以是任何类型的，默认是个 ObjectId 对象

由于 ObjectId 中保存了创建的时间戳，所以你不需要为你的文档保存时间戳字段，你可以通过 getTimestamp 函数来获取文档的创建时间:

> var newObject = ObjectId()
> newObject.getTimestamp()
ISODate("2017-11-25T07:21:10Z")

ObjectId 转为字符串

> newObject.str
5a1919e63df83ce79df8b38f

字符串
BSON 字符串都是 UTF-8 编码。

基本使用操作

操作数据库

创建数据库： use dbname，必须插入数据后数据库才真正创建
查看数据库： show dbs
删除数据库： db.dropDatabase()

操作集合

创建非固定集合： db.createCollection("COLLECTION_NAME")
创建固定集合： db.createCollection("COLLECTION_NAME", { capped : true, autoIndexId : true, size : 6142800, max : 10000 } )

查看集合： show collections
清空集合： db.COLLECTION_NAME.remove({}) ，过时了，需要接db.repairDatabase()来释放磁盘空间
如删除集合下全部文档：db.COLLECTION_NAME.deleteMany({})
删除 status 等于 A 的全部文档：db.COLLECTION_NAME.deleteMany({ status : "A" })
删除 status 等于 D 的一个文档：db.COLLECTION_NAME.deleteOne( { status: "D" } )
删除集合： db.COLLECTION_NAME.drop()

操作文档

插入文档： db.COLLECTION_NAME.insert(document)
插入文档： db.COLLECTION_NAME.save(document)，若document中没有携带_id参数功能与insert操作相同，若携带_id参数，则变成更新操作。
批量插入文档：db.COLLECTION_NAME.insertMany([document,document1])
更新文档：db.COLLECTION_NAME.update( <query>, <update>, { upsert: <boolean>, multi: <boolean>, writeConcern: <document> } )

参数	说明
query	update的查询条件，类似sql update查询内where后面的
update	update的对象和一些更新的操作符（如inc...）等，也可以理解为sql update查询内set后面的
upsert	可选，这个参数的意思是，如果不存在update的记录，是否插入objNew,true为插入，默认是false，不插入
multi	可选，mongodb 默认是false,只更新找到的第一条记录，如果这个参数为true,就把按条件查出来多条记录全部更新
writeConcern	可选，抛出异常的级别

例子：

db.col.update({'title':'MongoDB 教程'},{$set:{'title':'MongoDB'}},{multi:true})

查看文档： db.COLLECTION_NAME.find(query, projection)

用格式化方式显示结果，使用的是 pretty() 方法
条件查询： db.COLLECTION_NAME.find()
例子：

【and】
----------------------------------------------------------------
db.col.find({key1:value1, key2:value2}).pretty()
  ↓↓
Select * from col WHERE key1=value1 AND key2=value2
----------------------------------------------------------------

【or】
----------------------------------------------------------------
db.col.find({$or: [{key1: value1}, {key2:value2}]}).pretty()
  ↓↓
Select * from col WHERE key1=value1 OR key2=value2
----------------------------------------------------------------

【等于】
----------------------------------------------------------------
db.col.find({key1:value1}).pretty()
  ↓↓
Select * from col WHERE key1=value1
----------------------------------------------------------------

【小于】
----------------------------------------------------------------
db.col.find({key1:{$lt:value1}}).pretty()
  ↓↓
Select * from col WHERE key1<value1
----------------------------------------------------------------

【小于等于】
----------------------------------------------------------------
db.col.find({key1:{$lte:value1}}).pretty()
  ↓↓
Select * from col WHERE key1 <= value1
----------------------------------------------------------------

【大于】
----------------------------------------------------------------
db.col.find({key1:{$gt:value1}}).pretty()
  ↓↓
Select * from col WHERE key1 > value1
----------------------------------------------------------------

【大于等于】
----------------------------------------------------------------
db.col.find({key1:{$gte:value1}}).pretty()
  ↓↓
Select * from col WHERE key1 >= value1
----------------------------------------------------------------

【不等于】
----------------------------------------------------------------
db.col.find({key1:{$ne:value1}}).pretty()
  ↓↓
Select * from col WHERE key1 != value1
----------------------------------------------------------------

其他例子：

【and + or】 
----------------------------------------------------------------
db.col.find({"likes": {$gt:50}, $or: [{"by": "aaa"},{"title": "MongoDB 教程"}]}).pretty() 
类似于 
Select * from col where likes>50 AND (by = 'aaa' OR title = 'MongoDB 教程')
----------------------------------------------------------------

【> + <】
----------------------------------------------------------------
db.col.find({likes : {$lt :200, $gt : 100}})
类似于 
Select * from col where likes>100 AND  likes<200;
----------------------------------------------------------------

按字段类型查询： db.COLLECTION_NAME.find({"title" : {$type : 'string'}}) 或 db.col.find({"title" : {$type : 2}})

skip(), limilt(), sort()三个放在一起执行的时候，执行的顺序是先 sort(), 然后是 skip()，最后是显示的 limit()。

skip与limit与排序查询：db.COLLECTION_NAME.find().skip(NUMBER1).limit(NUMBER2).sort({KEY:1}) 跳过NUMBER1条；查询NUMBER2条；1为升序，-1为降序。

操作索引

索引通常能够极大的提高查询的效率，如果没有索引，MongoDB在读取数据时必须扫描集合中的每个文件并选取那些符合查询条件的记录。
这种扫描全集合的查询效率是非常低的，特别在处理大量的数据时，查询可以要花费几十秒甚至几分钟，这对网站的性能是非常致命的。
索引是特殊的数据结构，索引存储在一个易于遍历读取的数据集合中，索引是对数据库表中一列或多列的值进行排序的一种结构。

创建索引：db.COLLECTION_NAME.createIndex(keys, options)

image.png

例子：

db.col.createIndex({open: 1, close: 1}, {background: true})

查看集合索引：db.COLLECTION_NAME.getIndexes()
查看集合索引大小： db.COLLECTION_NAME.totalIndexSize()
删除集合所有索引：db.COLLECTION_NAME.dropIndexes()
删除集合指定索引：db.COLLECTION_NAME.dropIndex("索引名称")

聚合

聚合管道（aggregate pipeline）

什么是聚合管道？

文档通过多级管道将会输出聚合结果；aggregate管道聚合方案使用的是mongodb内置的汇总操作，相对来说更为高效，在做mongodb数据聚合操作的时候优先推荐aggregate；aggregate能够通过索引来提升性能，并且有一系列的管道性能优化操作

聚合管道使用场景？

1.用于常用聚合操作
2.对聚合响应性能需要一定要求时（索引及组合优化）
3.aggregate 管道操作是在内存中完成的，有内存大小限制，处理数据集大小有限；

聚合管道使用限制？

1.当aggregate返回的结果集（指针或者结果集），当结果集中的单个文档超过16 MB命令会报错；该限制只适用于返回的结果集文档，在管道处理文档的过程中，这个文档很有可能会超过16MB；

2.如果使用aggregate不指定游标选项或存储集合中的结果，aggregate命令返回一个包涵于结果集的字段中的bson文件。如果结果集的总大小超过bson文件大小限制（16MB）该命令将产生错误；

3.管道处理阶段有内存限制最大不能超过100MB,超过这个限制会报错误；为了能够处理更大的数据集可以开启allowDiskUse选项，可以将管道操作写入临时文件；

聚合管道语法

db.collection.aggregate(pipeline, options)

聚合通道类似于shell命令中的管道：

image.png

例子：

image.png

常见管道操作介绍

常见管道表达式介绍

详见：管道表达式

image.png

aggressive例子

参考：aggressive相关例子

映射规约（MapReduce）

什么是MapReduce？

Map-reduce是处理聚合计算的另外一个方式；Map-reduce一般有两个阶段：一个阶段是处理单个文档，另一个阶段是将处理完的当前文档返回一个或者多个对象进入下一个文档处理方法中；

简单来说就是reduce文档结果集，通过reduce函数进行汇总；

Map-reduce使用惯用的javascript操作来做map和reduce操作，因此Map-reduce的灵活性和复杂性都会比aggregate pipeline更高一些，并且相对aggregate pipeline而言更消耗性能；

MapReduce使用场景

1.因为使用javascript灵活度高的特点，可以处理复杂聚合需求；
2.管道对数据的类型和结果的大小会有一些限制，对于一些简单的固定的聚集操作可以使用管道，但是对于一些复杂的、大量数据集的聚合任务还是使用MapReduce。

MapReduce语法

db.runCommand({ 
    mapreduce:<collection>, 
    map:<mapfunction>,
    reduce:<reducefunction>,
    [,query:<query filter object>]
    [,sort:<sorts the input objects using this key.Useful for optimization,like sorting by the emit key for fewer reduces>]
    [,limit:<number of objects to return from collection>]
    [,out:<see output options below>]
    [,keeptemp:<true|false>]
    [,finalize:<finalizefunction>]
    [,scope:<object where fields go into javascript global scope>]
    [, jsMode : boolean，default true]
    [,verbose:true]
});

例子：

image.png

说明：

1. MongoDB中使用emit函数向MapReduce提供Key/Value对。
2. Reduce函数接受两个参数：Key,emits. Key即为emit函数中的Key。 emits是一个数组，它的元素就是emit函数提供的Value。
3. Reduce函数的返回结果必须要能被Map或者Reduce重复使用，所以返回结果必须与emits中元素结构一致。
4. Map或者Reduce函数中的this关键字，代表当前被Mapping文档。

MongoDB中的MapReduce主要有以下几阶段：

• Map：把一个操作Map到集合中的某一个文档
• Shuffle: 根据Key分组对文档，并且为每个不同的Key生成一系列(>=1个)的值表(List of values)。
• Reduce: 处理值表中的元素，直到值表中只有一个元素。然后将值表返回到Shuffle过程，循环处理，直到每个Key只对应一个值表，并且此值表中只有一个元素，这就是MR的结果。
• Finalize：此步骤不是必须的。在得到MR最终结果后，再进行一些数据“修剪”性质的处理。
  另一个例子：
  
  image.png

聚合命令

比aggregate性能低，比Map-reduce灵活度低；但是可以节省几行javascript代码

• Group操作：mongodb2.2版本对于返回数据最多只包涵20000个元素，最多支持20000独立分组；对于超过20000的独立分组建议采用mapreduce；

• Count操作：db.collection.count()等效于db.collection.find().count(),在分布式集合中，会出现计算错误的情况，这个时候推荐使用aggregate；

• Distinct操作：可以使用索引；

嵌入与引用

t o d o

查询分析

t o d o

原子操作

t o d o

高级索引

t o d o

索引限制

t o d o

全文检索

t o d o

正则表达式

t o d o

部署运维

备份与恢复

t o d o

监控

t o d o

集群部署

单实例

这种配置只适合简易开发时使用，生产使用不行，因为单节点挂掉整个数据业务全挂

image.png

主从模式

如何解决单点故障？

使用mysql数据库时大家广泛用到，采用双机备份后主节点挂掉了后从节点可以接替主机继续服务。所以这种模式比单节点的高可用性要好很多。

image.png

多个从节点

一个数据库服务器又提供写又提供读，机器承载会出现瓶颈，如何解决？

写大概占总压力的20%，读大概占总压力的80%。压力主要来自于读。写放到主节点，读放到多个从节点。通过读写分离解决单台机器的承载瓶颈问题。

image.png

副本集模式

1.主节点挂了，如何自动切换？
2.主节点的读写压力过大如何解决？
3.从节点每个上面的数据都是对数据库全量拷贝，从节点压力会不会过大？
4.数据压力大到机器支撑不了的时候能否做到自动扩展？

mongoDB官方已经不建议使用主从模式了，替代方案是采用副本集的模式。

image.png

副本集中的副本节点在主节点挂掉后通过心跳机制检测到后，就会在集群内发起主节点的选举机制，自动选举一位新的主服务器。从而解决主节点自动切换问题。但其他问题还仍未解决。

副本集模式，读写分离

2.主节点的读写压力过大如何解决？
3.从节点每个上面的数据都是对数据库全量拷贝，从节点压力会不会过大？
4.数据压力大到机器支撑不了的时候能否做到自动扩展？

1、设置读写分离需要先在副本节点SECONDARY 设置 setSlaveOk。
2、在程序中设置副本节点负责读操作，如下代码：

public class TestMongoDBReplSetReadSplit {
 
        public static void main(String[] args) {
 
               try {
                     List<ServerAddress> addresses = new ArrayList<ServerAddress>();
                     ServerAddress address1 = new ServerAddress("192.168.1.136" , 27017);
                     ServerAddress address2 = new ServerAddress("192.168.1.137" , 27017);
                     ServerAddress address3 = new ServerAddress("192.168.1.138" , 27017);
                     addresses.add(address1);
                     addresses.add(address2);
                     addresses.add(address3);
 
                     MongoClient client = new MongoClient(addresses);
                     DB db = client.getDB( "test" );
                     DBCollection coll = db.getCollection( "testdb" );
 
 
                     BasicDBObject object = new BasicDBObject();
                     object.append( "test2" , "testval2" );
 
                      //读操作从副本节点读取
                     ReadPreference preference = ReadPreference. secondary();
                     DBObject dbObject = coll.findOne(object, null , preference);
 
                     System. out .println(dbObject);
 
 
              } catch (Exception e) {
                     e.printStackTrace();
              }
       }
}

image.png

副本集模式，解决数据同步压力

3.从节点每个上面的数据都是对数据库全量拷贝，从节点压力会不会过大？
4.数据压力大到机器支撑不了的时候能否做到自动扩展？

我们一般有主节点、副本节点2个节点储存数据就够用了，但是为了能够完成选举，节点数量必须是奇数个，而又不想加大数据同步的压力。这时我们考虑使用仲裁节点，来避免多余的数据同步。

仲裁节点即投票节点，其本身并不包含数据集，且也无法晋升为主节点。

image.png

分片模式

4.数据压力大到机器支撑不了的时候能否做到自动扩展？

原来的1个T的数据，将其分成多个分片存储。每个分片，进行副本备份。

image.png

image.png

上图中主要有如下所述三个主要组件：

• Shard:用于存储实际的数据块，实际生产环境中一个shard server角色可由几台机器组个一个replica set承担，防止主机单点故障。

• Config Server:mongod实例，存储所有数据库元信息（路由、分片）的配置，分片服务器和数据路由信息。

• Query Routers:前端路由，客户端由此接入，且让整个集群看上去像单一数据库，前端应用可以透明使用。

使用3台机器，每台机器部署路由服务，配置服务，分片、副本节点、仲裁节点。

• 路由服务高可用
• 配置服务高可用
• 分片、副本、仲裁不在同一台机器上

物理部署图：

image.png

优化部署图：

其中的两台读写的机器比较繁忙，而仲裁的机器闲置资源。

调整服务负载更加均衡

image.png

其他问题：MongoDB客户端连接时，可以通过添加多个连接地址连接，但是这种机制，其内部原理是选择一个ping 最快的机器来作为所有请求的入口，这相当于没有做负载均衡。

所以路由服务Mongos可以为其添加负载均衡器，来解决此问题。

设计模式

嵌入还是引用？

标准化范式

1nf：行列原子化。2nf：列内数据不冗余。3nf：各属性与主键为直接关系。

原则

为了原子性、独立性，使用嵌入。（查询模式常见固定）
为了灵活性，使用引用。（查询模式多种、无法预计）
为了潜在高引用数量不超出限制时，使用引用。
为了多对多关系，使用引用。

智能运营系统

系统简述

从数据存储设备中获取原始数据，根据数据生成可操作的报告（理想情况数据能够实时导入），批处理生成各个层面的报告。

精简记录信息

image.png

大多数情况下，没必要记录事件日志中的所有数据。

image.png

BSON的ObjectId字段暗含了创建时间，进一步精简。

image.png

实践

Java Spring Data MongoDB使用

添加maven依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-mongodb</artifactId>
        </dependency>

通过aggravate计算访问次数

    /**
     * 获取访问次数
     * @return
     */
    private Long getVisitNum() {
        //pipeline
        Long from = DateUtils.get00TimeMillis() - MongoOperatorEnum.TIME_SIXDAY_MSEC.getValue();
        Criteria criteria = Criteria.where("last_visit").gt(from);   //条件：近7日

        MatchOperation match = Aggregation.match(criteria);
        GroupOperation group = Aggregation.group().sum("hit").as("cnt");
        Aggregation agg = Aggregation.newAggregation(match, group);

        //execute
        List<Document> ret = mongoTemplate.aggregate(agg, MongoOperatorEnum.COLL_NAME_STATUS_DAILY.getName(), Document.class).getMappedResults();

        if(ret.iterator().hasNext()){
            Object cnt = ret.iterator().next().get("cnt");
            return Double.valueOf(String.valueOf(cnt)).longValue();
        }

        return 0L;
    }

其中Aggregation.group().sum("hit").as("cnt");的group中没有添写字段，表示所有数据分为一组。添写字段即按照字段分组。

通过aggregate进行分页查询

    /**
     * 组装查询条件
     *
     * @param req
     * @param ruleData
     * @return
     */
    private Map<String, Aggregation> makeCondition(SenAppRequest req, List<SensitiveDataDiscernEntity> ruleData) {
        Map<String, Aggregation> ret = new HashMap<String, Aggregation>();

        //match
        Long from = DateUtils.get00TimeMillis() - MongoOperatorEnum.TIME_SIXDAY_MSEC.getValue();
        Criteria criteria = Criteria.where("last_visit").gt(from);

        if (Objects.nonNull(req.getHostname())) {
            //条件：模糊匹配应用名称
            Pattern pattern = Pattern.compile(new StringBuilder("^.*").append(req.getHostname()).append(".*$").toString(), Pattern.CASE_INSENSITIVE);
            criteria.and("hostname").regex(pattern);
        }

        MatchOperation match = Aggregation.match(criteria);

        //group4page
        GroupOperation groupA4page = Aggregation.group("url")
                .first("hostname").as("hostname")
                .sum("hit.daily").as("total_count");

        GroupOperation groupB4page = Aggregation.group("hostname")
                .count().as("url_count")
                .sum("total_count").as("total_count");

        for (SensitiveDataDiscernEntity rule : ruleData) {
            String key = MongoOperatorEnum.SUFFIX_RUlE_FIELD.getName() + rule.getId() + "";
            groupA4page = groupA4page.sum(key + ".daily").as(key);
            groupB4page = groupB4page.sum(key).as(key);
        }

        //4page
        SortOperation sort = Aggregation.sort(Sort.Direction.DESC, "total_count");
        SkipOperation skip = Aggregation.skip((req.getCurrentPage() - 1) * req.getPageSize());
        LimitOperation limit = Aggregation.limit(req.getPageSize());

        //group4count
        GroupOperation group4count = Aggregation.group("hostname");
        CountOperation count = Aggregation.count().as("count");

        Aggregation aggPage = Aggregation.newAggregation(match, groupA4page, groupB4page, sort, skip, limit);
        Aggregation aggCount = Aggregation.newAggregation(match, group4count, count);

        ret.put("count", aggCount);
        ret.put("page", aggPage);

        return ret;
    }

流量表里，一个hostname里面包含多个url。
Aggregation aggPage = Aggregation.newAggregation(match, groupA4page, groupB4page, sort, skip, limit);这里使用了两个group命令，第一个group对细粒度的url字段进行分组，求出聚合结果每个url的访问次数。接着第二个group对粗粒度的hostname字段进行分组，求出每个hostname对应了多少个不重复的url，并求出每个hostname的总访问次数。注意这里的分组和通常的按先粗、再细分组思路完全相反。

通过aggregate求7天折线图

    /**
     * 近七天 某应用的 敏感数据访问 折线图
     * 返回7个点，没有的补0
     *
     * @return
     */
    @Override
    public String getSenVisitLineChartByApp(SenAppRequest req) {
        //pipeline
        Long from = DateUtils.get00TimeMillis() - MongoOperatorEnum.TIME_SIXDAY_MSEC.getValue();
        Criteria criteria = Criteria.where("last_visit").gt(from);  //条件：近7日
        if (Objects.nonNull(req.getAppName())) {
            criteria.and("hostname").is(req.getAppName());         //条件：某应用
        }

        MatchOperation match = Aggregation.match(criteria);
        ProjectionOperation project = Aggregation.project()
                .and("hit.daily").as("cnt")
//                .and(DateOperators.DateToString.dateOf("create_time").toString("%Y-%m-%d")).as("day");
                .and(StringOperators.Substr.valueOf("_id").substring(0, 8)).as("day");
        GroupOperation group = Aggregation.group("day").sum("cnt").as("amount");
        LimitOperation limit = Aggregation.limit(7);
        Aggregation agg = Aggregation.newAggregation(match, project, group, limit);

        //execute
        List<Document> ret = mongoTemplate.aggregate(agg, MongoOperatorEnum.COLL_NAME_STATUS_DAILY.getName(), Document.class).getMappedResults();

        //声明封装结果所需变量
        LinkedHashMap<String, Object[]> map = new LinkedHashMap<String, Object[]>();
        ArrayList<LinkedHashMap<String, Object[]>> list = new ArrayList<LinkedHashMap<String, Object[]>>();
        LinkedHashMap<String, Integer> tmpMap = new LinkedHashMap<String, Integer>();

        //放一周的初始值
        for (int i = 6; i >= 0; i--) {
            tmpMap.put(getPastDate(i), 0);
        }

        if (Objects.nonNull(ret) && ret.size() > 0) {
            for (Document ele : ret) {
                StringBuilder time = new StringBuilder(String.valueOf(ele.get("_id")));
                time.insert(4, "-").insert(7, "-");
//                String time = String.valueOf(ele.get("_id"));
                Integer amount = Double.valueOf(String.valueOf(ele.get("amount"))).intValue();
                tmpMap.put(time.toString(), amount);
            }
        }

        map.put("createTime", tmpMap.keySet().toArray());
        map.put("totalCount", tmpMap.values().toArray());
        list.add(map);
        return EncapsulationResultUtil.setResult(list);
    }

//.and(DateOperators.DateToString.dateOf("create_time").toString("%Y-%m-%d")).as("day");
ProjectionOperation project = Aggregation.project().and("hit.daily").as("cnt").and(StringOperators.Substr.valueOf("_id").substring(0, 8)).as("day");
其中的这一句，用于将原始数据映射出我们需要的字段，再根据这个拼出来的字段分组。这里我们截取了“_id”字符串的一部分形如（20190101）当做日期，再根据它分组。

通过mapreduce 命令查询数据

这里讲MapReduce命令封装了一个实体类。

    /**
     * 计算敏感应用列表信息
     */
    @Override
    public void caculateSenAppList(boolean isReal) {
        //组装sql
        Document cmd = this.assembleSql4senAppList(isReal);
        //mapreduce
        mongoTemplate.executeCommand(cmd);
    }

    /**
     * 组装 敏感应用列表 map-reduce CMD
     *
     * @return Document
     */
    private Document assembleSql4senAppList(boolean isReal) {
        //局部变量
        List<SensitiveDataDiscernEntity> list = sensitiveDataDiscernService.querySensitiveDataDiscernManager();
        Long to = 0L;
        Long from = 0L;
        if (isReal) {
            //今天0点 ~ 当前时间
            from = DateUtils.get00TimeMillis();
            to = System.currentTimeMillis();
        } else {
            //6天前0点 ~ 今天0点
            from = DateUtils.get24TimeMillis() - MongoOperatorEnum.TIME_1WEEK_MSEC.getValue();
            to = DateUtils.get00TimeMillis();
        }

        StringBuilder m = new StringBuilder();
        StringBuilder r = new StringBuilder();
        StringBuilder f = new StringBuilder();
        StringBuilder part1_m = new StringBuilder();
        StringBuilder part1_r = new StringBuilder();
        StringBuilder part2_r = new StringBuilder();
        StringBuilder part3_r = new StringBuilder();
        StringBuilder part1_f = new StringBuilder();

        //查询条件
        Query query = new Query();

        Criteria criteria = Criteria.where("create_time").ne(null).lte(to).gt(from)
                .and("is_sensitive").is(1);
        query.addCriteria(criteria);

        //拼装m、r
        if (Objects.nonNull(list) && list.size() > 0) {
            for (SensitiveDataDiscernEntity i : list) {
                StringBuilder name = new StringBuilder(MongoOperatorEnum.SUFFIX_RUlE_FIELD.getName()).append(i.getId());
                part1_m.append(",").append(name).append(":this.").append(name);
                part1_r.append("var ").append(name).append("=0;");
                part2_r.append(name).append("+=val.").append(name).append(";");
                part3_r.append(",").append(name).append(":").append(name);
                part1_f.append(",").append(name).append(":rs.").append(name);
            }
            m.append("function () {emit({hostname:this.hostname}, {url:this.url,total_count:this.total_count")
                    .append(part1_m)
                    .append("})}");
            r.append("function(key, values){")
                    .append("var total_count=0;")
                    .append("var url_count=0;")
                    .append(part1_r)
                    .append("var n = {},r=[];")
                    .append("values.forEach(function(val){")
                    .append("total_count+=val.total_count;")
                    .append(part2_r)
                    .append("if (!n[val.url]){")
                    .append("n[val.url]=true;")
                    .append("r.push(val.url);")
                    .append("}});")
                    .append("url_count=r.length;")
                    .append("return {url_count:url_count,total_count:total_count")
                    .append(part3_r)
                    .append("};}");
            f.append("function(key,rs){return {url_count:rs.url_count,total_count:rs.total_count,create_time:new Date().valueOf()")
                    .append(part1_f)
                    .append("}}");
        }

        //input collection
        String input = MongoOperatorEnum.COLL_NAME_BROKER.getName();

        //out
        StringBuilder out = new StringBuilder();

        if(isReal){
            out.append(MongoOperatorEnum.COLL_NAME_SEN_APP_REAL.getName());
        }else {
            out.append(MongoOperatorEnum.COLL_NAME_SEN_APP.getName());
        }

        //返回
        return SimpleMR.builder()
                .mapReduce(input)
                .map(m.toString())
                .reduce(r.toString())
                .query(query.getQueryObject())
                .finalize(f.toString())
                .out(out.toString())
                .build().toDocument();
    }

通过将mapreduce需要的变量封装成一个实体类，再转换为一个Document，继而使用mongoTemplate.executeCommand(cmd);执行，这种方法非常灵活。

批量插入

            List<JSONObject> list = new ArrayList<>();
            JSONObject json = JSON.parseObject(JSON.toJSONString(bd));
            list.add(json );
            BulkOperations ops = mongoTemplate.bulkOps(
                    BulkOperations.BulkMode.UNORDERED,"coll_name");
            ops.insert(list);
            ops.execute();

批量插入的效率高于逐条插入。

mapreduce api

        MapReduceOptions mrOps=new MapReduceOptions();
        mrOps.outputDatabase("coll_name"); //output集合
        mrOps.outputTypeReplace(); //output策略，替换
        mrOps.verbose(true); //详细日志
        mrOps.javaScriptMode(true); //map-reduce中间过程不转bson
        mrOps.finalizeFunction("xxx"); //finalize函数

        MapReduceResults<Document> mrr = mongoTemplate.mapReduce(
                    mr.getQuery(), mr.getCollectionName(),
                    mr.getMap(), mr.getReduce(),mrOps, Document.class);

其中需要注意，mrOps.outputDatabase("coll_name"); //output集合这句话并不生效，不能输出到另一个集合bug。
另，timestamp比较方便进行时间的比较，推荐在mongodb中存储timestamp类型。当timestamp类型从mongodb中取出到java中时会变为科学记数法从而不能变成需要的13位字符串。通过下面方式转换一下：

                SimpleDateFormat sdf =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                BigDecimal ts = new BigDecimal(Double.valueOf(String.valueOf(i.get("last_visit"))));
                String last_visit = sdf.format(new Date(ts.longValue()));

GO SDK使用

t o d o

相关FAQ

1.同一台服务器上同时开始两个或以上mongod实例，发生内存不足而异常退出

现象

1>用top命令查看系统占用内存的情况 top -p $(pidof mongod)，发现mongod占用了8G内存的35.6%。在服务器上运行两个mongod进程，很容易导致mongod异常退出
2>内存不足引发bulk_write_exception

image.png

原因

MongoDB使用的是内存映射存储引擎，它会把磁盘IO操作转换成内存操作，如果是读操作，内存中的数据起到缓存的作用，如果是写操作，内存还可以把随机的写操作转换成顺序的写操作，总之可以大幅度提升性能。MongoDB并不干涉内存管理工作，而是把这些工作留给操作系统的虚拟缓存管理器去处理，这样的好处是简化了MongoDB的工作，但坏处是你没有方法很方便的控制MongoDB占多大内存，事实上MongoDB会占用所有能用的内存，所以最好不要把别的服务和MongoDB放一起。

MongDB把内存管理交给操作系统，缺了就和操作系统要，多了并不还，通常一个mongod服务要占用一台主机的一半内存以上，2个mongod服务就容易导致内存不足。

解决方案

能够收回内存的方式：
1>重启mongod，或者调用 db.runCommand({closeAllDatabases:1})来清除内存
2>使用Linux命令清除缓存中的数据：echo 3 > /proc/sys/vm/drop_caches
3>使用 --wiredTigerCacheSizeGB设置内存大小

cacheSizeGB 指的就是Cache size，包括数据和索引。Mongod本身使用内存如连接池堆栈以及sorting buffer等都是额外的，不会被统计到这个数字里面。如果索引在内存，查询冷数据取决于你的IO能力。如果IO latency很低，系统也没有是高负载，那响应时间应该是毫秒级的区别。但是如果查询很频繁，又涉及到很多范围、批量查询，IOPS又跟不上，那就有可能到几百几千毫秒都有可能。

2.map过程映射出一条记录后，不走reduce过程，直接返回。

To Be Continued

参考：

怎样学 MongoDB?
搭建高可用MongoDB集群
map reduce例子
spring data mongodb 分页
MongoDB如何实现实时排名
MongoDB进阶模式设计
MongoDB数据建模小案例：物联网时序数据库建模
MongoDB数据建模小案例：多列数据结构
MongoDB实战-面向文档的数据(找到最合适的数据建模方式)