mongodb的中文文档地址官方地址
1、增加单条数据
db.collection.insert() 将一个或多个文档插入到集合中。实例：
db.getCollection('test').insert([ {"name":"adas","sex":"sada","passion":"fdaf"}, {"name":"萨达","sex":"阿斯顿撒","passion":"给对方"} ])2、增加多条数据
db.collection.insertMany()将多个文档插入到集合中。 db.getCollection('test').insertMany([ {"name":"王八蛋","sex":"芽儿呦","passion":"kkk"}, {"name":"大多数","sex":"阿斯顿撒","passion":"阿斯顿撒"} ])3、增加单条数据
db.collection.insertOne() 将单个文档插入集合中。 db.getCollection('test').insert( {"name":"滚滚滚","sex":"三国杀","passion":"哦咯"} )1、删除多个
db.collection.deleteMany() 删除所有与指定过滤器匹配的文档。 # 删除所有传入{}文档，即 db.collection.deleteMany({})2、删除一个
db.collection.deleteOne() 即使多个文档可能与指定过滤器匹配，也最多删除一个与指定过滤器匹配的文档。3、删除一个或多个
db.collection.remove() 删除单个文档或与指定过滤器匹配的所有文档。1、更新一个
db.collection.updateOne() 即使多个文档可能与指定的过滤器匹配，也最多更新一个与指定的过滤器匹配的文档2、更新多个
db.collection.updateMany() 更新所有与指定过滤器匹配的文档。3、替换一个
db.collection.replaceOne() 即使多个文档可能与指定过滤器匹配，也最多替换一个与指定过滤器匹配的文档。4、更新单个或多个
db.collection.update() 更新或替换与指定过滤器匹配的单个文档，或更新与指定过滤器匹配的所有文档。实例： db.test.find({})
示例：
db.characters.bulkWrite( [ { insertOne : { "document" : { "_id" : 4, "char" : "Dithras", "class" : "barbarian", "lvl" : 4 } } }, { insertOne : { "document" : { "_id" : 5, "char" : "Taeln", "class" : "fighter", "lvl" : 3 } } }, { updateOne : { "filter" : { "char" : "Eldon" }, "update" : { $set : { "status" : "Critical Injury" } } } }, { deleteOne : { "filter" : { "char" : "Brisbane"} } }, { replaceOne : { "filter" : { "char" : "Meldane" }, "replacement" : { "char" : "Tanys", "class" : "oracle", "lvl" : 4 } } } ] )1、先实现客户端连接
import pymongo from contextlib import contextmanager class MongoDB: def __init__(self, host, port, db): self.host = host self.port = port self.db = db @contextmanager def get_client(self): mg = pymongo.MongoClient(self.host, self.port) try: dbs = mg.get_database(self.db) yield dbs except Exception as e: print(e) finally: mg.close()2、插入单条或者多条数据
向集合test中插入一个文档。如果集合不存在，这个操作将创建一个新的集合。
client.insert_one() #插入单条数据 client.insert_many([]) #插入多条数据代码示例：
mg = MongoDB(host='127.0.0.1', port=27017, db="test_demo") with mg.get_client() as mogo: client = mogo.get_collection("test") # 插入数据 res = client.insert_many([ {"name":"呜啊还能", "get":"吸星大法", "skills":["葵花宝典", "五零龙首"]}, {"name":"阿斯顿马丁", "get":"彩虹灯车道", "skills":["秋明山", "22号公路"]}, ]) print(res.inserted_ids) # 返回值 [ObjectId('5f923ced30a8593eacc573e5'), ObjectId('5f923ced30a8593eacc573e6')] # 插入单条数据 res = client.insert_one( {"name": "asfgsdf ", "get": "sadsad", "skills": ["jhm", "l.asdas"]}, ) print(res.inserted_id) # 返回值res是一个对象3、查询操作 你可以通过find()方法产生一个查询来从MongoDB的集合中查询到数据。MongoDB中所有的查询条件在一个集合中都有一个范围。
查询可以返回在集合中的所有数据或者只返回符合筛选条件（filter）或者标准（criteria）的文档。你可以在文档中指定过滤器或者标准，并作为参数传递给find()方法。
代码示例：
with mg.get_client() as mogo: client = mogo.get_collection("test") res = [json.dumps(x) for x in list(client.find({}, {"_id": 0}))] # 返回的本是游标对象，这里格式化转为列表形式 print(res) # 结果注意： 由于里面的ObjectId 不能json话，就和时间格式一样，需要转为字符串或另外的json可识别的类型即可 Object of type ObjectId is not JSON serializable
所以在查询的时候不显示_id字段。
# 如果不想显示某个字段 {'_id':0}加入都查询语句后，即可。如上示例高级查询
条件操作符
result = task_cache_collection.find({"loop_count": {"$gt": 12}}) #$gt 大于 result = task_cache_collection.find({"loop_count": {"$lt": 12}}) #$lt 小于 result = task_cache_collection.find({"loop_count": {"$gte": 12}}) #$gte 大于等于 result = task_cache_collection.find({"loop_count": {"$lte": 12}}) #$lte 小于等于 result = task_cache_collection.find({"loop_count": {"$lte": 12, "$gte": 2}}) #多条件 2 < loop_count < 12$all 匹配所有
这个操作符跟 SQL 语法的 in 类似，但不同的是, in 只需满足( )内的某一个值即可, 而$all 必 须满足[ ]内的所有值，例如:
result = list(task_cache_collection.find({"loop_count": {"$all": [12, 12]}}, {"_id": 0})) # 匹配ok result = list(task_cache_collection.find({"loop_count": {"$all": [12, 1]}}, {"_id": 0})) # 匹配不到$exists 判断字段是否存在
查询所有存在 age 字段的记录 task_cache_collection.find({age: {$exists: true}}); 查询所有不存在 age 字段的记录 task_cache_collection.find({age: {$exists: false}});$ne 不等于
result = task_cache_collection.find({"loop_count": {"$ne": 12}}) #$ne 不等于 result = task_cache_collection.find({"loop_count": 12}) #: 等于$in 包含
result = list(task_cache_collection.find({"task_type": {"$in": wakeup_types}}, {"_id": 0}))$nin 不包含
result = list(task_cache_collection.find({"task_type": {"$nin": wakeup_types}}, {"_id": 0}))$size 数组元素个数
对于{name: 'David', age: 26, favorite_number: [ 6, 7, 9 ] }记录 匹配 db.users.find({favorite_number: {$size: 3}}); 不匹配 db.users.find({favorite_number: {$size: 2}});正则表达式匹配 $regex
task_cache_collection.find({"task_type":{'$regex' : ".*atmosphere.*"}})count 查询记录条数
skip 限制返回记录的起点
result = list(task_cache_collection.find({"task_type": {"$in": wakeup_types} if task_type in wakeup_types else task_type}, {"_id": 0}).sort('update_time', asc).skip(skip_page).limit(count))$or
result = list(all_tasks.find({"$or": [{"age": 23}, {"age": 27}]}, {"_id": 0}))聚合查询 $match
# 使用aggregate聚合查询，已字段的某个几个查询并分组， 先过滤在分组 res = list(all_tasks.aggregate( [{"$match": {"task_type": {"$in": ["LinkTask", "LinkTask2", "LinkAppTask"]}}}, {"$group": {"_id": "$task_type", "task_detail": { "$push": {"task_type": "$tak_type", "task_uuid": "$task_uuid", "task_name": "$task_name"}}}}])) # 如果要显示数据中的目标字段而不是全部，使用$push，如上，要显示全部{"$push": "$$ROOT"}, 这个查询默认是隐藏字段的$group 使用分组时需要注意的是，以哪个字段进行分组时必须使用_id，否则会报错
# 使用aggregate聚合查询，已字段的某个几个查询并分组， 先过滤在分组 res = list(all_tasks.aggregate( [{"$match": {"task_type": {"$in": ["LinkTask", "LinkTask2", "LinkAppTask"]}}}, {"$group": {"_id": "$task_type", "task_detail": { "$push": {"task_type": "$tak_type", "task_uuid": "$task_uuid", "task_name": "$task_name"}}}}])) # 如果要显示数据中的目标字段而不是全部，使用$push，如上，要显示全部{"$push": "$$ROOT"}, 这个查询默认是隐藏字段的$sample
# 随机显示某几条数据 res = list(all_tasks.aggregate([{"$sample": {"size": 2}}]))$lookup 多表查询
源数据
db.orders.insert([ { "_id" : 1, "item" : "almonds", "price" : 12, "quantity" : 2 }, { "_id" : 2, "item" : "pecans", "price" : 20, "quantity" : 1 }, { "_id" : 3 } ]) db.inventory.insert([ { "_id" : 1, "sku" : "almonds", description: "product 1", "instock" : 120 }, { "_id" : 2, "sku" : "bread", description: "product 2", "instock" : 80 }, { "_id" : 3, "sku" : "cashews", description: "product 3", "instock" : 60 }, { "_id" : 4, "sku" : "pecans", description: "product 4", "instock" : 70 }, { "_id" : 5, "sku": null, description: "Incomplete" }, { "_id" : 6 } ]) # $lookup 多表联合查询 all_tasks = db.get_collection(TEST_ORDER) # 场景一：查找订单对应的商品 res = list(all_tasks.aggregate([{ "$lookup": { "from": "inventory", # 要连接查询的集合 "localField": "item", # 源集合的连接查询字段 "foreignField": "sku", # 目标集合和源集合连接查询的字段 "as": "inventory_docs" # 目标集合中查出的数据，list形态 } }]))结果
注意 1. 这个示例中，一共输出了三个文档，在没有再次聚合（$match）的条件下， 这个输出文档数量是以输入文档的数量来决定的（由order来决定），而不是以 被Join的集合（inventory）文档数量决定。 2. 在此需要特别强调的是输出的第三个文档。在源库中原文档没有要比较的列 （即item值不存在，既不是Null值也不是值为空），此时和被Join集合比较， 如果被Join集合中比较列也恰好为NUll或不存在的值，此时，判断相等， 即会把被Join集合中比较列为NUll或值不存在文档吸收进来。 { "api_code": 2000, "data": [ { "_id": 1.0, "inventory_docs": [ { "_id": 1.0, "description": "product 1", "instock": 120.0, "sku": "almonds" } ], "item": "almonds", "price": 12.0, "quantity": 2.0 }, { "_id": 2.0, "inventory_docs": [ { "_id": 4.0, "description": "product 4", "instock": 70.0, "sku": "pecans" } ], "item": "pecans", "price": 20.0, "quantity": 1.0 }, { "_id": 3.0, "inventory_docs": [ { "_id": 5.0, "description": "Incomplete", "sku": null }, { "_id": 6.0 } ] } ], "msg": "success!", "request": "GET /v1/task/get_all_link_task/" }新数据
db.orders.insert({ "_id" : 1, "item" : "MON1003", "price" : 350, "quantity" : 2, "specs" :[ "27 inch", "Retina display", "1920x1080" ], "type" : "Monitor" }) db.inventory.insert({ "_id" : 1, "sku" : "MON1003", "type" : "Monitor", "instock" : 120,"size" : "27 inch", "resolution" : "1920x1080" }) db.inventory.insert({ "_id" : 2, "sku" : "MON1012", "type" : "Monitor", "instock" : 85,"size" : "23 inch", "resolution" : "1280x800" }) db.inventory.insert({ "_id" : 3, "sku" : "MON1031", "type" : "Monitor", "instock" : 60,"size" : "23 inch", "display_type" : "LED" }) res = list(all_tasks.aggregate([ { "$unwind": "$specs", # 拆分数组匹配 }, { "$lookup": { "from": "inventory", # 要连接查询的集合 "localField": "specs", # 源集合的连接查询字段 "foreignField": "size", # 目标集合和源集合连接查询的字段 "as": "inventory_docs" # 目标集合中查出的数据，list形态 } }, { "$match": {"inventory_docs": {"$ne": []}} }, { "$project": {"inventory_docs": 1, "_id":0} # 限制显示那些字段 } ]))结果
{ "api_code": 2000, "data": [ { "inventory_docs": [ { "_id": 1.0, "instock": 120.0, "resolution": "1920x1080", "size": "27 inch", "sku": "MON1003", "type": "Monitor" } ] } ], "msg": "success!", "request": "GET /v1/task/get_all_link_task/" }4、更新操作 你可以使用update_one()和update_many方法更新集合中的文档。update_one()方法一次更新一个文档。使用update_many()方法可以更新所有符合条件的文档。方法接受以下三个参数：
一个筛选器，可以对符合条件的文档进行更新。 一个指定的修改语句 自定义更新时的操作参数
更新单个update_one
result = db.restaurants.update_one( {"name": "Juni"}, { "$set": { "cuisine": "American (New)" }, "$currentDate": {"lastModified": True} } )更新多个update_many
result = db.restaurants.update_many( {"address.zipcode": "10016", "cuisine": "Other"}, { "$set": {"cuisine": "Category To Be Determined"}, "$currentDate": {"lastModified": True} } )替换replace_one 要替换整个文档（除了_id字段），将一个完整的文档作为第二个参数传给update()方法。替代文档对应原来的文档可以有不同的字段。在替代文档中，你可以忽略_id字段因为它是不变的。如果你包含了_id字段，那它必须和原文档的值相同。
mg = MongoDB(host='127.0.0.1', port=27017, db="test_demo") with mg.get_client() as mogo: client = mogo.get_collection("test") res = client.replace_one({"name": "asfgsdf "}, {"adc":"lll", "kkk":"oko"}) print(res.matched_count)5、删除操作 你可以使用delete_one()以及delete_many()方法从集合中删除文档。方法需要一个条件来确定需要删除的文档。
mg = MongoDB(host='127.0.0.1', port=27017, db="test_demo") with mg.get_client() as mogo: client = mogo.get_collection("test") res = client.delete_many({"adc":"lll"}) print(res.deleted_count)6、嵌套查询 数据库里的数据是这样的 mongo命令行的查询语句：
db.getCollection('task_products').find({'products.product_category':'监听设备'})即，外部字段.内部字段。这里查询的全部，数据格式基本要一致。
Mongo的锁是表级锁，mysql(innoDB引擎)和oracle是行级锁。 由于建立索引是需要花费时间的，而MongoDB的索引也是B+树，都知道索引也是占用磁盘空间的。建立索引就像给一新书建立目录，需要花费时间，并且期间mongo不允许别人来操作。
1、建索引导致数据库阻塞
mongo的查询是将所有数据读取到内存中进行查询的，而查询字段如果建立了索引就会先读取索引到内存，然后找到对应的查询区域
MongoDB 库级锁的问题，建索引就是一个容易引起长时间写锁的问题，MongoDB 在前台建索引时需要占用一个写锁（而且不会临时放弃），如果集合的数据量很大，建索引通常要花比较长时间，特别容易引起问题。
初看起来库级锁在大并发环境下有严重的问题，但是 MongoDB 依然能够保持大并发量和高性能，这是因为 MongoDB 的锁粒度虽然很粗放，但是在锁处理机制和关系数据库锁有很大差异，主要表现在：
MongoDB 没有完整事务支持，操作原子性只到单个 document 级别，所以通常操作粒度比较小； MongoDB 锁实际占用时间是内存数据计算和变更时间，通常很快； MongoDB 锁有一种临时放弃机制，当出现需要等待慢速 IO 读写数据时，可以先临时放弃，等 IO 完成之后再重新获取锁。
解决的方法很简单，MongoDB 提供了两种建索引的访问，一种是 background 方式，不需要长时间占用写锁，另一种是非 background 方式，需要长时间占用锁。使用 background 方式就可以解决问题。 例如，为超大表 posts 建立索引， 千万不用使用
db.posts.ensureIndex({user_id: 1})而应该使用后台运行方式，{background:1}
db.posts.ensureIndex({user_id: 1}, {background: 1})具体示例：
db.getCollection('task_products').find({'products.product_category':'监听设备'}, {backgroud:1}) db.getCollection('task_products').find({'products.product_category':'监听设备'})索引的增删改查
# 我这里使用pymongo增删改查 # 创建索引 # res = list(all_tasks.create_index([("age", -1)])) # 显示集合所有索引 # res = list(all_tasks.list_indexes()) # print(res) # 修改索引 又名重建索引 不建议使用 res = all_tasks.reindex() print(res) # 删除索引 这里只能全部删除索引 # res = all_tasks.drop_indexes() # print(res) # # 查询 # s = time.time() # res = list(all_tasks.find({"age": 25}, {"_id": 0})) # sec = time.time() - s # return APISuccess(results={"data": res, "sec": sec})使用索引之后查询耗时： 使用索引之前查询耗时 使用了索引之后确实查询效率提高了不少 复合索引 explain()查看是否正确命中索引
1 为什么要执行explain，什么时候执行 explain的目的是将mongo的黑盒操作白盒化。 比如查询很慢的时候想知道原因。 2 explain的三种模式 2.1 queryPlanner 不会真正的执行查询，只是分析查询，选出winning plan。 2.2 executionStats 返回winning plan的关键数据。 executionTimeMillis该query查询的总体时间。 2.3 allPlansExecution 执行所有的plans。 通过explain("executionStats")来选择模式，默认是第一种模式。 3 queryPlanner分析 3.1 namespace 本次所查询的表。 3.2 indexFilterSet 是否使用partial index，比如只对某个表中的部分文档进行index。 3.3 parsedQuery 本次执行的查询 3.4 winning plan 3.4.1 stage COLLSCAN：全表扫描 IXSCAN：索引扫描 FETCH：根据索引去检索指定document SHARD_MERGE：将各个分片返回数据进行merge SORT：表明在内存中进行了排序 LIMIT：使用limit限制返回数 SKIP：使用skip进行跳过 IDHACK：针对_id进行查询 # 创建复合索引 res = list(all_tasks.create_index([("age", -1), ("name", 1)])) # 查询使用索引 res = all_tasks.find({"age": 25, "name": "UUKS"}, {"_id": 0}).explain() #命中索引 res = all_tasks.find({"name": "UUKS", "age": 25}, {"_id": 0}).explain() #命中索引 res = all_tasks.find({"name": "UUKS"}, {"_id": 0}).explain() #无法命中 全表扫描 res = all_tasks.find({"age": 25}, {"_id": 0}).explain() # 最左原则正确命中无论复合索引，多复杂，如果没有最左侧索引，都无法正确使用索引。复合索引遵循最左匹配原则
1. 建立复合索引，索引中两字段前后顺序与查询条件字段在数量一致的情况下，顺序不影响使用索引查询。 2.当复合索引中的字段数量与查询条件字段数量不一致情况下，以复合索引字段前置优先规则。适合建立索引的字段
基于合理的数据库设计，经过深思熟虑后为表建立索引，是获得高性能数据库系统的基础。而未经合理分析便添加索引，则会降低系统的总体性能。索引虽然说提高了数据的访问速度，但同时也增加了插入、更新和删除操作的处理时间。 是否要为表增加索引、索引建立在那些字段上，是创建索引前必须要考虑的问题。解决此问题的一个比较好的方法，就是分析应用程序的业务处理、数据使用，为经常被用作查询条件、或者被要求排序的字段建立索引。基于优化器对SQL语句的优化处理，我们在创建索引时可以遵循下面的一般性原则： （1）为经常出现在关键字order by、group by、distinct后面的字段，建立索引。 在这些字段上建立索引，可以有效地避免排序操作。如果建立的是复合索引，索引的字段顺序要和这些关键字后面的字段顺序一致，否则索引不会被使用。 （2）在union等集合操作的结果集字段上，建立索引。其建立索引的目的同上。 （3）为经常用作查询选择的字段，建立索引。 （4）在经常用作表连接的属性上，建立索引。 （5）考虑使用索引覆盖。对数据很少被更新的表，如果用户经常只查询其中的几个字段，可以考虑在这几个字段上建立索引，从而将表的扫描改变为索引的扫描。 除了以上原则，在创建索引时，我们还应当注意以下的限制： （1）限制表上的索引数目。 对一个存在大量更新操作的表，所建索引的数目一般不要超过3个，最多不要超过5个。索引虽说提高了访问速度，但太多索引会影响数据的更新操作。 （2）不要在有大量相同取值的字段上，建立索引。 在这样的字段（例如：性别）上建立索引，字段作为选择条件时将返回大量满足条件的记录，优化器不会使用该索引作为访问路径。 （3）避免在取值朝一个方向增长的字段（例如：日期类型的字段）上，建立索引；对复合索引，避免将这种类型的字段放置在最前面。 由于字段的取值总是朝一个方向增长，新记录总是存放在索引的最后一个叶页中，从而不断地引起该叶页的访问竞争、新叶页的分配、中间分支页的拆分。此外，如果所建索引是聚集索引，表中数据按照索引的排列顺序存放，所有的插入操作都集中在最后一个数据页上进行，从而引起插入“热点”。 （4）对复合索引，按照字段在查询条件中出现的频度建立索引。 在复合索引中，记录首先按照第一个字段排序。对于在第一个字段上取值相同的记录，系统再按照第二个字段的取值排序，以此类推。因此只有复合索引的第一个字段出现在查询条件中，该索引才可能被使用。 因此将应用频度高的字段，放置在复合索引的前面，会使系统最大可能地使用此索引，发挥索引的作用。 （5）删除不再使用，或者很少被使用的索引。 表中的数据被大量更新，或者数据的使用方式被改变后，原有的一些索引可能不再被需要。数据库管理员应当定期找出这些索引，将它们删除，从而减少索引对更新操作的影响。在MongoDB中，操作单个文档（document）是原子性的；但是，涉及到多个文档的操作，也就是常说的“多文档事务”，是非原子性的。由于document可以设计的非常复杂，包含多个“内嵌的”文档，因此单个文档的原子性为很多实际场景提供了必要的支持。 尽管单文档原子操作很强大，但在很多场景下依然需要多文档事务。当执行一个由几个顺序操作组成的事务时，可能会出现某些问题，例如：
原子性：如果某个操作失败了，在同一个事务中前面的操作回滚到最初的状态（即，要么全做，要么全部不做）。 一致性：如果发生了严重故障将事务中断（比如：网络、硬件故障），数据库必须能够恢复到一致的状态。 在需要多文档事务的场景中，你可以实现两阶段提交来完成场景需求。两阶段提交可以保证数据的一致性， 如果发生错误，可以恢复到事务开始之前的状态。在事务执行过程中，无论发生什么情况都可以还原到数据和状态 的准备阶段。1.MongoDB需要4.0版本+
2.需要自己搭建MongoDB复制集，单个mongodb server 不支持事务。
事务原理：mongodb的复制至少需要两个节点。其中一个是主节点，负责处理客户端请求，其余的都是从节点，负责复制主节点上的数据。mongodb各个节点常见的搭配方式为：一主一从、一主多从。主节点记录在其上的所有操作oplog，从节点定期轮询主节点获取这些操作，然后对自己的数据副本执行这些操作，从而保证从节点的数据与主节点一致。
