mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-22 02:53:09 +00:00
feature: mongo 增加 bulkWrite (#2058)
mongo8.0 新增跨集合更新api: bulkWrite(相对单一更新 性能大幅提升) btw: 低量级数据建议继续选择单一更新(insert/update等 ) 应用场景: 1. 聚合数据更新 2. 日志db跨集合写入
This commit is contained in:
@@ -88,6 +88,38 @@ local function __parse_addr(addr)
|
||||
return host, tonumber(port)
|
||||
end
|
||||
|
||||
local function werror(r)
|
||||
local ok = (r.ok == 1 and not r.writeErrors and not r.writeConcernError and not r.errmsg)
|
||||
|
||||
local err
|
||||
if not ok then
|
||||
if r.writeErrors then
|
||||
err = r.writeErrors[1].errmsg
|
||||
elseif r.writeConcernError then
|
||||
err = r.writeConcernError.errmsg
|
||||
else
|
||||
err = r.errmsg
|
||||
end
|
||||
end
|
||||
return ok, err, r
|
||||
end
|
||||
|
||||
local function wbulkerror(r)
|
||||
local ok, err = werror(r)
|
||||
local cursor
|
||||
if (r.nErrors and r.nErrors ~= 0) and r.cursor then
|
||||
cursor = {}
|
||||
err = err and err .. ";" or ""
|
||||
for i, v in ipairs(r.cursor.firstBatch) do
|
||||
if v.ok ~= 1 then
|
||||
cursor[#cursor+1] = i
|
||||
err = err .. v.errmsg .. ";"
|
||||
end
|
||||
end
|
||||
end
|
||||
return ok, err, r, cursor
|
||||
end
|
||||
|
||||
local auth_method = {}
|
||||
|
||||
local function mongo_auth(mongoc)
|
||||
@@ -197,6 +229,61 @@ function mongo_client:runCommand(...)
|
||||
return self.admin:runCommand(...)
|
||||
end
|
||||
|
||||
local function filter(tbl, key, value)
|
||||
if value == nil then return end
|
||||
tbl[#tbl+1] = key
|
||||
tbl[#tbl+1] = type(value) == "table" and bson_encode(value) or value
|
||||
end
|
||||
|
||||
local bulkWrite = {}
|
||||
bulkWrite.insert = function(nsindex, doc)
|
||||
return bson_encode_order("insert", nsindex, "document", bson_encode(doc.insert))
|
||||
end
|
||||
|
||||
bulkWrite.delete = function(nsindex, doc)
|
||||
local args = {"delete", nsindex, "filter", bson_encode(doc.filter), "multi", doc.multi or false}
|
||||
filter(args, "hit", doc.hit)
|
||||
filter(args, "collation", doc.collation)
|
||||
return bson_encode_order(table.unpack(args))
|
||||
end
|
||||
|
||||
--update仅支持原子操作更新/聚合管道更新 详情阅读官方文档
|
||||
--arrayFilters 过滤方式需要遍历encode 暂时不支持
|
||||
bulkWrite.update = function(nsindex, doc)
|
||||
local args = {"update", nsindex, "filter", bson_encode(doc.filter), "updateMods", bson_encode(doc.update),
|
||||
"multi", doc.multi or false, "upsert", doc.upsert or false}
|
||||
filter(args, "hit", doc.hit)
|
||||
filter(args, "constants", doc.constants)
|
||||
filter(args, "collation", doc.collation)
|
||||
return bson_encode_order(table.unpack(args))
|
||||
end
|
||||
|
||||
--bulkWrite 依赖 mongo8.0x
|
||||
--忽略let,cursor,writeConcern 参数
|
||||
---@param datas table 待写入数据
|
||||
--- - op string 写入数据方式 参见:bulkWrite
|
||||
--- - collection string 数据写入目标, 用于构建nsInfo; 注意:采用 空间.集合方式
|
||||
--- - ... 操作具体参数 insert采用:insert填充document, update采用 update 填充updateMods
|
||||
--- 示例:{op = "insert", collection = "log.online", insert = {count=0}}
|
||||
--- {op = "update", collection = "log.online", update = {["$set"] = {count = 2}}, upsert = true}}
|
||||
function mongo_client:bulkWrite(datas, comment, ordered, verify, errorsOnly)
|
||||
local ops, ns, map = {}, {}, {}
|
||||
for i, v in ipairs(datas) do
|
||||
if not map[v.collection] then
|
||||
map[v.collection] = #ns --ops 索引从0开始
|
||||
ns[#ns+1] = bson_encode({ns = v.collection})
|
||||
end
|
||||
local f = assert(bulkWrite[v.op], v.op)
|
||||
ops[i] = f(map[v.collection], v)
|
||||
end
|
||||
local args = {"bulkWrite", 1, "ops", ops, "nsInfo", ns}
|
||||
filter(args, "ordered", ordered) --是否有序执行 默认有序
|
||||
filter(args, "bypassDocumentValidation", verify) --是否验证 默认验证
|
||||
filter(args, "comment", comment) --日志跟踪注释
|
||||
filter(args, "errorsOnly",errorsOnly) -- 仅返回错误信息
|
||||
return wbulkerror(self:runCommand(table.unpack(args)))
|
||||
end
|
||||
|
||||
function auth_method:auth_mongodb_cr(user,password)
|
||||
local password = md5.sumhexa(string.format("%s:mongo:%s",user,password))
|
||||
local result= self:runCommand "getnonce"
|
||||
@@ -355,22 +442,6 @@ function mongo_collection:insert(doc)
|
||||
self.database:send_command("insert", self.name, "documents", {bson_encode(doc)})
|
||||
end
|
||||
|
||||
local function werror(r)
|
||||
local ok = (r.ok == 1 and not r.writeErrors and not r.writeConcernError and not r.errmsg)
|
||||
|
||||
local err
|
||||
if not ok then
|
||||
if r.writeErrors then
|
||||
err = r.writeErrors[1].errmsg
|
||||
elseif r.writeConcernError then
|
||||
err = r.writeConcernError.errmsg
|
||||
else
|
||||
err = r.errmsg
|
||||
end
|
||||
end
|
||||
return ok, err, r
|
||||
end
|
||||
|
||||
function mongo_collection:safe_insert(doc)
|
||||
local r = self.database:runCommand("insert", self.name, "documents", {bson_encode(doc)})
|
||||
return werror(r)
|
||||
|
||||
Reference in New Issue
Block a user