什么是 MongoDB?
MongoDB 和 Redis 一样均为 key-value 存储系统,它具有以下特点:
MongoDB 与 MySQL 性能比较
MongoDB应用范围和限制
MongoDB 适用范围如下:
pymongo是python访问MongoDB的模块,使用该模块,我们定义了一个操作MongoDB的类PyMongoClient,包含了连接管理、集合管理、索引管理、增删改查、文件操作、聚合操作等方法。
# ---------------------------------------------------------------------------------------------------
# PyMongoClient(host='localhost', port='27017', db='test', user=None, passwd=None, loop=5, rate=8)
# ---------------------------------------------------------------------------------------------------
PyMongoClient.SetDatabase(db, user, passwd) # 设置(修改)当前数据库
PyMongoClient.CloseConnection() # 关闭连接
PyMongoClient.Logout() # 注销用户
PyMongoClient.GetStatus()# 获取MogoDB服务器的状态
PyMongoClient.IsMongos() # 判断是否是MongoS
PyMongoClient.GetDateTime()# 获取MongoDB服务器的当前时间(需要权限支持)
# ---------------------------------------------------------------------------------------------------
PyMongoClient.GetCollections() # 获取当前数据库的全部集合
PyMongoClient.CreateCollection(collection)# 创建集合
PyMongoClient.DropCollection(collection)# 删除集合
# ---------------------------------------------------------------------------------------------------
PyMongoClient.IndexInformation(collection)# 获取集合的索引信息
PyMongoClient.EnsureIndex(collection, key_or_list)# 检查索引是否存在,若不存在,则创建索引
PyMongoClient.CreateIndex(collection, key_or_list)# 创建索引
PyMongoClient.DropIndex(collection, key=None)# 删除索引,key=None时删除全部索引(_id除外)
# ---------------------------------------------------------------------------------------------------
PyMongoClient.InsertDoc(collection, data)# data为字典时,单条插入,data为列表时,批量插入
PyMongoClient.RemoveDoc(collection, docFilter=None)# 删除文档,docFilter=None时删除集合的全部文档
PyMongoClient.UpdateDoc(collection, docFilter, data, modifier=None)# 更新文档,支持使用$inc/$set/$unset等修改器
PyMongoClient.UpsertDoc(collection, docFilter, data)# 如果文档不存在,则插入文档;如果文档存在,则更新文档
PyMongoClient.GetDoc(collection, docFilter=None, colFilter=None)# 返回单个文档
PyMongoClient.CountDoc(collection, docFilter=None)# 返回集合或查询的文档总数
PyMongoClient.GetCursor(collection, docFilter=None, colFilter=None)# 返回多个文档的游标
PyMongoClient.CountCursor(cursor) # 返回游标的文档总数
PyMongoClient.SortCursor(cursor, col_or_list, director='ASC')# 游标排序,默认升序,取值ASC/DESC
PyMongoClient.SubCursor(cursor, limit, skip=0)# 游标截取
# ---------------------------------------------------------------------------------------------------
PyMongoClient.Aggregate(collection, pipleline) # 聚合
PyMongoClient.RunCommand(collection, cmdObj) # 运行数据库命令
# ---------------------------------------------------------------------------------------------------
PyMongoClient.Str2ObjectId(id_str)# id字符串转id对象
PyMongoClient.ObjectId2Str(id_obj)# id对象转id字符串
PyMongoClient.GetBinaryFromFile(sourceFile) # 读文件,返回二进制内容
PyMongoClient.SaveBinaryToFile(binary, targetFile) # 将二进制内容保存为文件
# ---------------------------------------------------------------------------------------------------
PyMongoClient.PutFile(localFilePath, dbFileName=None) # 将文件保存到GridFS并返回FileId
PyMongoClient.GetFile(fileId, localFilePath)# 将文件从GridFS取出,并保存到文件中
PyMongoClient.GetFilesCursor(docFilter=None, colFilter=None)# 取得文件信息游标
PyMongoClient.DeleteFile(fileId) # 删除GridFS中的文件
class PyMongoClient():
def __init__(self, host='localhost', port=27017, db='test', user=None, passwd=None, loop=5, rate=8):
self.loop = loop # 数据库失去连接后,尝试执行数据库操作的次数
self.rate = float(rate) # 数据库失去连接后,尝试执行数据库操作的时间间隔,首次尝试的间隔是rate的倒数,以后间隔时间增倍
try:
self.conn = pymongo.MongoClient(host, int(port))
self.SetDatabase(db, user, passwd)
except Exception, errMsg:
raise Exception(errMsg)
# ---------------------------------------------------------------------------------------------------
def SetDatabase(self, db, user, passwd):
# 设置(修改)当前数据库
self.db = self.conn[db]
if user and passwd:
if not self.db.authenticate(user, passwd):
raise Exception(u'数据库权限验证失败!')
def CloseConnection(self):
# 关闭数据库连接
self.conn.close()
def Logout(self):
# 注销用户
self.db.logout()
def GetStatus(self):
# 获取MogoDB服务器的状态
return self.db.last_status()
def IsMongos(self):
# 判断是否是MongoS
return self.conn.is_mongos
def GetDateTime(self):
# 获取MongoDB服务器的当前时间(需要权限支持,若无权限,则返回本地时间)
for i in range(self.loop):
try:
return self.db.eval("return new Date();")
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
except Exception, e:
return datetime.datetime.now()
raise Exception(u'重连数据库失败!')
class PyMongoClient():
def GetCollections(self):
# 获取当前数据库的全部集合
for i in range(self.loop):
try:
return self.db.collection_names()
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def CreateCollection(self, collection):
# 在当前数据库内创建新的集合
for i in range(self.loop):
try:
self.db.create_collection(collection)
return
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
except pymongo.errors.CollectionInvalid:
return
raise Exception(u'重连数据库失败!')
def DropCollection(self, collection):
# 删除当前数据库内名为collection的集合
for i in range(self.loop):
try:
self.db.drop_collection(collection)
return
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
class PyMongoClient():
def IndexInformation(self, collection):
# 获取索引信息
for i in range(self.loop):
try:
return self.db[collection].index_information().keys()
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def EnsureIndex(self, collection, key_or_list):
# 检查索引是否存在,若不存在,则创建索引,若存在,返回None
# list参数形如:[('start_time', pymongo.ASCENDING), ('end_time', pymongo.ASCENDING)]
for i in range(self.loop):
try:
self.db[collection].ensure_index(key_or_list)
return
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def CreateIndex(self, collection, key_or_list):
# 创建索引(推荐使用EnsureIndex)
for i in range(self.loop):
try:
self.db[collection].create_index(key_or_list)
return
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def DropIndex(self, collection, key=None):
# 删除索引,key=None时删除全部索引(_id除外)
for i in range(self.loop):
try:
if key:
self.db[collection].drop_index(key)
else:
self.db[collection].drop_indexes()
return
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
class PyMongoClient():
def InsertDoc(self, collection, data):
# data为字典时,单条插入,data为列表时,批量插入。批量上限20W
# 单条插入时返回单个id对象,批量插入时,返回id对象列表
for i in range(self.loop):
try:
return self.db[collection].insert(data, manipulate=True, save=False, check_keys=True)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def RemoveDoc(self, collection, docFilter=None):
# 删除文档,docFilter=None时删除集合collection的全部文档
for i in range(self.loop):
try:
return self.db[collection].remove(docFilter)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def UpdateDoc(self, collection, docFilter, data, modifier=None):
# 更新文档,docFilter为更新对象的查找条件,data为更新数据,可以使用$inc/$set/$unset等修改器
for i in range(self.loop):
try:
if modifier:
return self.db[collection].update(docFilter, {modifier:data}, multi=True)
else:
return self.db[collection].update(docFilter, data, multi=True)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def UpsertDoc(self, collection, docFilter, data):
# 如果文档不存在,则插入文档;如果文档存在,则更新文档
for i in range(self.loop):
try:
return self.db[collection].update(docFilter, data, True)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def GetDoc(self, collection, docFilter=None, colFilter=None, sort=None):
# 返回单个文档
for i in range(self.loop):
try:
if colFilter:
return self.db[collection].find_one(docFilter, colFilter, sort=sort)
else:
return self.db[collection].find_one(docFilter, sort=sort)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def CountDoc(self, collection, docFilter=None):
# 返回集合或查询的文档总数
for i in range(self.loop):
try:
return self.db[collection].find(docFilter).count()
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def GetCursor(self, collection, docFilter=None, colFilter=None):
# 返回多个文档的游标
for i in range(self.loop):
try:
if colFilter:
return self.db[collection].find(docFilter, colFilter).batch_size(100)
else:
return self.db[collection].find(docFilter).batch_size(100)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def CountCursor(self, cursor):
# 返回游标的文档总数
for i in range(self.loop):
try:
return cursor.count()
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def SortCursor(self, cursor, col_or_list, director='ASC'):
# 游标排序,默认ASCENDING(升序),取值ASC/DESC
# col_or_list,列名或者是由(列名,方向)组成的列表
if isinstance(col_or_list, list):
args = []
for col in col_or_list:
if col[1] == 'ASC':
args.append((col[0],pymongo.ASCENDING))
else:
args.append((col[0],pymongo.DESCENDING))
for i in range(self.loop):
try:
return cursor.sort(args) # cursor.sort([("UserName",pymongo.ASCENDING),("Email",pymongo.DESCENDING)])
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
else:
if director == 'ASC':
director = pymongo.ASCENDING
else:
director = pymongo.DESCENDING
for i in range(self.loop):
try:
return cursor.sort(col_or_list, director) # director取值:pymongo.ASCENDING(升序)、pymongo.DESCENDING(降序)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def SubCursor(self, cursor, limit, skip=0):
# 截取游标
for i in range(self.loop):
try:
if skip:
return cursor.skip(skip).limit(limit)
else:
return cursor.limit(limit)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
class PyMongoClient():
def GetBinaryFromFile(self, sourceFile):
# 读文件,返回二进制内容
# 适用于在文档中直接保存小于16M的小文件,若文件较大时,应使用GridFS
try:
fp = open(sourceFile,'rb')
return bson.Binary(fp.read())
except:
return False
finally:
fp.close()
def SaveBinaryToFile(self, binary, targetFile):
# 将二进制内容保存为文件
try:
fp = open(targetFile,'wb')
fp.write(binary)
return True
except:
return False
finally:
fp.close()
def Str2ObjectId(self, id_str):
return bson.ObjectId(id_str)
def ObjectId2Str(self, id_obj):
return str(id_obj)
def PutFile(self, localFilePath, dbFileName=None):
'''
向GridFS中上传文件,并返回文件ID
@localFilePath 本地文件路径
@dbFileName 保存到GridFS中的文件名,如果为None则使用本地路径中的文件名
'''
fs = gridfs.GridFS(self.db)
fp = open(localFilePath, 'rb')
if dbFileName == None:
dbFileName = os.path.split(localFilePath)[1]
id = fs.put(fp,filename=dbFileName, chunkSize=4*1024*1024)
fp.close()
return id
def GetFile(self, fileId, localFilePath=None):
'''
根据文件ID从GridFS中下载文件
@fileId 文件ID
@localFilePath 要保存的本地文件路径
'''
if isinstance(fileId, str):
fileId = self.Str2ObjectId(fileId)
fs = gridfs.GridFS(self.db)
if localFilePath:
fp = open(localFilePath, 'wb')
try:
fp.write(fs.get(fileId).read())
return True
except:
return False
finally:
fp.close()
else:
try:
return fs.get(fileId).read()
except:
return False
def GetFilesCursor(self, docFilter=None, colFilter=None):
'''
取得GridFS中文件的游标
可以进行过滤或检索的字段名有
_id 文件ID
filename 文件名
length 文件大小
md5 md5校验码
chunkSize 文件块大小
uploadDate 更新时间
'''
return self.GetCursor('fs.files', docFilter=docFilter, colFilter=colFilter)
def DeleteFile(self, fileId):
'''
根据文件ID从GridFS中删除文件
@fileId 文件ID
'''
fs = gridfs.GridFS(self.db)
fs.delete(fileId)
class PyMongoClient():
def Aggregate(self, collection, pipleline):
# 聚合
# pipleline是一个由筛选、投射、分组、排序、限制、跳过等一系列构件组成管道队列
for i in range(self.loop):
try:
return self.db[collection].aggregate(pipleline)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
def RunCommand(self, collection, cmdObj):
# 运行数据库命令
# if cmdObj is a string, turns it into {cmdObj:1}
for i in range(self.loop):
try:
return self.db[collection].runCommand(cmdObj)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败!')
更高的写入负载:
高可用性:
MongoDB应用举例:
mkdir tools # 存放安装包
mkdir training # 存放安装目录
tar -zxvf mongodb-linux-x86_64-enterprise-rhel70-3.4.10.tgz -C ~/training/
vi ~/.bash_profile
# .bash_profile
MONGODB_HOME=/root/training/mongodb-linux-x86_64-enterprise-rhel70-3.4.10
export MONGODB_HOME
PATH=$MONGODB_HOME/bin:$PATH
export PATH
source ~/.bash_profile # 生效环境变量, 使在任何位置都可以使用MongoDB命令
安装MongoDB依赖rpm包
yum install -y net-snmp* # 安装net-snmp :
yum install -y cyrus* # 安装cyrus :
启动MongoDB
mkdir -p /data/db
mongod --dbpath /data/db # 指定MongoDB数据库文件存储路径(从3.2版本后,MongoDB的默认的数据引擎:wiredTiger)
mongod --dbpath /data/db1 --storageEngine=mmapv1 # 早期:内存映射,可以指定参数 --storageEngine=mmapv1
使用MongoDB Web的控制台:需要在启动MongoDB的时候,指定参数: --httpinterface
mongod --dbpath /data/db1 --storageEngine=mmapv1 --httpinterface
测试MongoDB shell
mongo # 进入mongo shell环境
show dbs # 查看所有db
use mydemo # 使用我们的数据库(如果没有这个数据库会自动创建)
db.test1.insert({id:1,"name":"Tom"}) # 在test1表中插入一条数据(如果没有这个表会自动创建)