Merge branch 'develop' into tr/ob

* develop: 扩容 redis 和 log服务监听再前增加池子拆分网关清除引导功能顺便也清除关卡解锁触发的强制引导

Merge branch 'develop' into tr/ob
* develop: 扩容 redis 和 log服务监听再前增加池子拆分网关清除引导功能顺便也清除关卡解锁触发的强制引导
zhouhaihai
2 parents 445aae2c 3f604f2e
Showing 20 changed files with 174 additions and 62 deletions Show diff stats
robot/robot.conf
robot/robot.lua
robot/robot_main.lua
robot/robot_pool.lua
robot/start.lua
robot/unitTest/item.lua
src/actions/GmAction.lua
src/agent.lua
src/main.lua
src/models/RoleLog.lua
src/services/agent_ctrl.lua
src/services/dbseed.lua
src/services/globald.lua
src/services/httpweb.lua
src/services/logd.lua
src/services/pvpd.lua
src/services/redisd.lua
src/services/watchdog.lua
src/shared/logproxy.lua
src/shared/redisproxy.lua
 root = "./"
-thread = 8
+thread = 4
 logger = nil
 harbor = 0
-start = "start"	-- main script
+start = "robot_main"	-- main script
 bootstrap = "snlua bootstrap"	-- The service for bootstrap
 lua_path = root .."skynet/lualib/?.lua;"..root.."robot/?.lua;"..root.."src/?.lua"
@@ -16,6 +16,7 @@ local eventListener = {}
 local ignoreListener = {
 	["Role.updateProperty"] = function(data)
+		if not client.role then return end
 		local msg = MsgPack.unpack(data)
 		for _, one in pairs(msg) do
 			client.role[one.key] = one.newValue
@@ -80,7 +81,7 @@ function triggerListener(listener, data)
 			ignoreListener[listener](data)
 		end
 	else
-		log(listener .. " no handle!!!")
+		-- log(listener .. " no handle!!!")
 	end
 end
@@ -103,8 +104,8 @@ function sendServer(actionCode, bin)
 end
 function requestServer(actionCode, bin, callback, isKeep)
-	sendServer(actionCode, bin)
 	addListener(actionCode, callback, isKeep)
+	sendServer(actionCode, bin)
 end
 local function heartBeat()
@@ -0,0 +1,21 @@
+local skynet = require "skynet"
+local config = require "robot_config"
+
+local preOnlineCount = 10
+
+
+skynet.start(function()
+	local need = math.ceil(config.online / preOnlineCount)
+	local inpre = math.ceil(config.inpre / need)
+	local idRange = math.ceil(config.max / need)
+	local curId = 1
+	local poold = skynet.newservice("robot_pool")
+	local pooldObj = skynet.call(poold, "lua", "start", config.online + preOnlineCount)
+
+	for i = 1, need do
+		local start = skynet.newservice("start")
+		skynet.send(start, "lua", "start", poold, pooldObj, preOnlineCount, curId, curId + idRange - 1, inpre)
+		curId = curId + idRange
+	end
+	skynet.exit()	
+end)
 \ No newline at end of file
@@ -0,0 +1,46 @@
+--[[
+	deque 为了减少锁竞争，尽量保证 a服务 push；b服务 pop
+]]
+local skynet = require "skynet"
+local deque = require "deque"
+
+local CMD = {}
+local factory
+
+local PRE_FEED_COUNT = 5
+local dead = 0
+-- agent死亡，通知poold补充，当累计到5个agent时，马上生成5个agent放入poold中
+-- 当然这里也可以写得更复杂，参考redis落地规则
+function CMD.feed()
+	dead = dead + 1
+	if dead == PRE_FEED_COUNT then
+		dead = 0
+		for i=1, PRE_FEED_COUNT do
+			factory:push(skynet.newservice("robot"))
+		end
+	end
+end
+
+-- 系统启动，生成count个agent放入双端队列中
+function CMD.start(count)
+	factory, addr = deque.new(count)
+
+	for i = 1, count do
+		factory:push(skynet.newservice("robot"))
+	end
+
+	return addr
+end
+
+local function __init__()
+	skynet.dispatch("lua", function (_, _, command, ...)
+		local f = CMD[command]
+		if command == "start" then
+			skynet.ret(skynet.pack(f(...)))
+			return
+		end
+		f(...)
+	end)
+end
+
+skynet.start(__init__)
@@ -5,13 +5,17 @@ local config = require &quot;robot_config&quot;
 local skynet = require "skynet"
 local netpack = require "skynet.netpack"
 local socketdriver = require "skynet.socketdriver"
+local deque = require "deque"
 local string_format = string.format
+local poold, pooldObj, onlineCount, startId, endId, inpre
+local factory
+
 local queue = {}
 local fd2serv = {}
 local id2fd = {}
 local MSG = {}
-local id_max = 0 -- robot id
+local id_max 
 function MSG.open( ... )
@@ -21,9 +25,11 @@ end
 function MSG.close(fd)
 	if fd2serv[fd] and not fd2serv[fd].closing then
 		fd2serv[fd].closing = true
+		print(fd2serv[fd].agent)
 		skynet.call(fd2serv[fd].agent, "lua", "exit")
-		log(string_format("logout %s", fd2serv[fd].id))
+		pcall(skynet.send, poold, "lua", "feed")
+		log(string_format("logout %s", fd2serv[fd].id))
 		id2fd[fd2serv[fd].id] = nil
 		fd2serv[fd] = nil
 	end
@@ -72,13 +78,14 @@ skynet.register_protocol {
 }
 local function add_robot()
-	local robot = skynet.newservice("robot")
+	local time = skynet.time()
+	local robot = factory:pop()
 	local fd = socketdriver.connect(config.host, config.port)
 	socketdriver.start(fd)
 	local id
-	if id_max >= config.max then
+	if id_max >= endId then
 		while true do
-			id = math.randomInt(1, config.max)
+			id = math.randomInt(startId, endId)
 			if not id2fd[id] then
 				break
 			end
@@ -89,9 +96,8 @@ local function add_robot()
 	end
 	fd2serv[fd] = {agent = robot, id = id}
 	id2fd[id] = fd
-	pcall(skynet.call, robot, "lua", "start", fd, id)
+	pcall(skynet.call, robot, "lua", "start", fd, id, prefix)
 	log(string_format("login %s", fd2serv[fd].id))
-
 	-- 定时下线
 	skynet.timeout(math.randomInt(config.online_time[1], config.online_time[2]) * 100, function()
 		MSG.close(fd)
@@ -106,8 +112,8 @@ local function online()
 	end
 	-- 及时补充人数
-	if curCount < config.online then
-		for i = curCount + 1, config.online do
+	if curCount < onlineCount then
+		for i = curCount + 1, onlineCount do
 			add_robot()
 		end
 	end
@@ -116,11 +122,15 @@ local function online()
 	skynet.timeout(100, online)
 end
-local function start()
+local function start(pooldp, pooldObj, onlineCountp, startIdp, endIdp, inprep)
 	log("start testing ...")
-	for i = 1, config.online, config.inpre do
-		for j = 1, config.inpre do
+	factory = deque.clone(pooldObj)
+	poold, onlineCount, startId, endId, inpre = pooldp, onlineCountp, startIdp, endIdp, inprep
+	id_max = startId - 1
+
+	for i = 1, onlineCount, inpre do
+		for j = 1, inpre do
 			add_robot()
 		end
 		skynet.sleep(100)
@@ -131,5 +141,10 @@ local function start()
 end
 skynet.start(function()
-	skynet.fork(start)
+	skynet.dispatch("lua", function (_, _, command, ...)
+		if command == "start" then
+			start(...)
+			return
+		end
+	end)
 end)
 \ No newline at end of file
@@ -14,7 +14,7 @@ end
 local taskMap = {
 	saleItem = {100},
 	openItem = {100},
-	openTimeBox = {100},
+	-- openTimeBox = {100},
 }
 function _M:task()
@@ -502,6 +502,11 @@ function _M.guide(role, pms)
 				end
 			end
 		end
+		for _, data in pairs(csvdb["guide_unlockCsv"]) do
+			if data.type == 0 or data.type == 3 then
+				str = str:setv(data.id,1)
+			end
+		end
 		role:updateProperty({field = "funcGuide", value = str})
 		role:mylog("gm_action", {desc = "sguide", key1 = pms.sender})
@@ -356,11 +356,6 @@ skynet.start(function()
 		return info
 	end)
-	redisd = skynet.localname(".redis")
-	if tonumber(skynet.getenv "logd") == 1 then
-		logd = skynet.localname(".log")
-	end
-
 	cs = queue()
 	pvpd = skynet.localname(".pvpcross")
@@ -2,23 +2,41 @@ local skynet = require &quot;skynet&quot;
 local max_client = tonumber(skynet.getenv("max_client"))
 local max_queue = tonumber(skynet.getenv("max_queue"))
+local work_count = tonumber(skynet.getenv("thread"))
+local use_logd = tonumber(skynet.getenv("logd"))
 skynet.start(function()
 	print("Server start")
 	skynet.newservice("debug_console", tonumber(skynet.getenv("debug_port")))
+
+	-- 启动redis
+	for i = 1, work_count do
+		local redisd = skynet.newservice("services/redisd", i)
+		skynet.call(redisd, "lua", "open", {
+			redishost = skynet.getenv("redis_host"),
+			redisport = tonumber(skynet.getenv("redis_port")),
+			redisdb = tonumber(skynet.getenv("redis_db")),
+			auth = skynet.getenv("redis_auth")
+		})
+	end
+
+	--启动log
+	if use_logd == 1 then
+		for i = 1, work_count do
+			local logd = skynet.newservice("services/logd", i)
+			skynet.call(logd, "lua", "open")
+		end
+	end
+
 	local httpd = skynet.newservice("services/httpweb", tonumber(skynet.getenv("httpweb_port")))
 	local watchdog = skynet.newservice("services/watchdog", max_client)
+
 	skynet.call(watchdog, "lua", "start", {
 		port = tonumber(skynet.getenv("server_port")),
 		maxclient = max_client + max_queue + 10,
 		httpd = httpd,
-
-		redishost = skynet.getenv("redis_host"),
-		redisport = tonumber(skynet.getenv("redis_port")),
-		redisdb = tonumber(skynet.getenv("redis_db")),
-		auth = skynet.getenv("redis_auth"),
 	})
 	skynet.exit()
 local serverId = skynet.getenv("servId")
 local server_id = (skynet.getenv("serverType") or "localtest") .. "_" .. serverId
+local logproxy = require "shared.logproxy"
 --[[
 	100	购买/兑换行为			
@@ -765,8 +766,7 @@ function RoleLog.bind(Role)
 				end
 			end
 		end
-		if not logd then return end
-		pcall(skynet.send, logd, "lua", "log", doc, "bi")
+		logproxy:log(doc, "bi")
 	end
 	function Role:logItems(itemId, before, after, log)
@@ -825,8 +825,7 @@ function RoleLog.bind(Role)
 			end
 		end
 		doc["@type"] = logType
-		if not logd then return end
-		pcall(skynet.send, logd, "lua", "log", doc, "log")
+		logproxy:log(doc, "log")
 	end
@@ -6,6 +6,7 @@ local xxtea = require &quot;xxtea&quot;
 local deque = require "deque"
 local datacenter = require "skynet.datacenter"
 local agent_queued = require "services.agent_queued"
+local logproxy = require "shared.logproxy"
 local pcall = pcall
 local string_format = string.format
@@ -139,11 +140,11 @@ function _M:check_agent_status()
 		end
 	end
-	if now >= next_log_time and now % 60 == 0 and logd then
+	if now >= next_log_time and now % 60 == 0 then
 		next_log_time = now + 60
 		local count = table_nums(self.u2f)
 		datacenter.set("onlineCount", count)
-		pcall(skynet.send, logd, "lua", "log", {["@type"] = "online", count = count}, "log")
+		logproxy:log({["@type"] = "online", count = count}, "log")
 	end
 end
@@ -34,8 +34,6 @@ local steps = {
 }
 skynet.start(function ()
-	redisd = skynet.localname(".redis")
-
 	redisproxy = require("shared.redisproxy")
 	local new = redisproxy:hsetnx("autoincrement_set", "server_start", os.date("%Y%m%d", skynet.timex())) == 1
@@ -190,7 +190,6 @@ local function __init__()
 			end
 		end
 	end)
-	redisd = skynet.localname(".redis")
 	skynet.register(".globald")
 end
@@ -53,12 +53,8 @@ end
 local CMD = require "actions.HttpAction"
 local function start()
-	redisd = skynet.localname(".redis")
 	globalCsv = csvdb["GlobalDefineCsv"]
-	if tonumber(skynet.getenv "logd") == 1 then
-		logd = skynet.localname(".log")
-	end
 	local listen_socket = socket.listen("0.0.0.0", port)
 	print("Listen web port " .. port)
@@ -2,12 +2,12 @@ local skynet = require &quot;skynet&quot;
 local queue = require "skynet.queue"
 local bson = require "bson"
 local socketdriver = require "skynet.socketdriver"
-
 local serverId = tonumber(skynet.getenv("servId"))
 require "shared.init"
 require "skynet.manager"
+local logdIdx = ...
 local table_insert = table.insert
 local pairs = pairs
 local ipairs = ipairs
@@ -15,6 +15,7 @@ local string_format = string.format
 local logId = 0
 local CMD, cs = {}
+local prefix = "wasteland" .. logdIdx .. "S" .. serverId ..  "C" 
 local logHandle = {
 	bi = {
@@ -33,7 +34,7 @@ local logHandle = {
 			doc["game_name"] = "wasteland"
 			doc["env"] = "cb"
 			doc["game_name_type"] = "guaji"
-			doc["log_id"] = "wastelandC" .. logId .. "S" .. serverId .. "T" ..  now
+			doc["log_id"] = prefix .. logId .. "T" ..  now
 			logId = (logId + 1) % 10000000
 		end
 	},
@@ -135,7 +136,7 @@ local function __init__()
 	end)
 	cs = queue()
-	skynet.register(".log")
+	skynet.register(".logd" .. logdIdx)
 end
 skynet.start(__init__)
@@ -289,7 +289,6 @@ end
 ------------------------------------------------------
 function CMD.start()
-	redisd = skynet.localname(".redis")
 	globalCsv = csvdb["GlobalDefineCsv"]
 	pvpInfo = require("models.Pvpd").new({key = "cross:pvpInfo"})
@@ -3,7 +3,7 @@ require &quot;skynet.manager&quot;
 local redis = require "skynet.db.redis"
 local db
-
+local idx = ...
 local command = {}
 function command.open(conf)
@@ -27,5 +27,5 @@ skynet.start(function()
 	skynet.info_func(function()
 		return skynet.stat("mqlen")
 	end)
-	skynet.register ".redis"
+	skynet.register(".redis" .. idx)
 end)
 \ No newline at end of file
 local skynet = require "skynet"
 require "skynet.manager"
-local redisproxy = require "shared.redisproxy"
 local socket = require "skynet.socket"
 local netpack = require "skynet.netpack"
 local datacenter = require "skynet.datacenter"
@@ -47,7 +46,7 @@ function SOCKET.data(fd, msg)
  	end
 end
-local use_logd = tonumber(skynet.getenv "logd")
+
 -- @desc: agent状态定时检测
 function check_agent_status()
@@ -65,11 +64,7 @@ end
 function CMD.start(conf)
 	skynet.call(gate_serv, "lua", "open" , conf)
-	skynet.call(redisd, "lua", "open", conf)
-
-	if use_logd == 1 then
-		skynet.call(logd, "lua", "open")
-	end
+	
 	skynet.call(pvpd, "lua", "start")
 	-- 开启agent状态检测定时器
 	check_agent_status()
@@ -110,17 +105,11 @@ skynet.start(function()
 		end
 	end)
 	skynet.register ".watchdog"
-	-- 数据库服务
-	redisd = skynet.newservice("services/redisd")
 	-- 提前加载好
 	csvdata.init()
 	print("launch csvdatad ...")
-	-- 日志服务
-	if use_logd == 1 then
-		logd = skynet.newservice("services/logd")
-	end
 	-- pvp 服务
 	pvpd = skynet.newservice("services/pvpd")
 	cluster.register("pvpd", pvpd)
@@ -0,0 +1,23 @@
+
+local skynet = require "skynet"
+local logd_count = tonumber(skynet.getenv("thread"))
+local use_logd = tonumber(skynet.getenv("logd"))
+
+
+local logd
+skynet.init(function()
+	if use_logd then
+		local idx = math.randomInt(1, logd_count)
+		logd = skynet.localname(".logd" .. idx)
+	end
+end)
+
+local logproxy = {}
+
+function logproxy:log(doc, logTo)
+	if use_logd and logd then
+		pcall(skynet.send, logd, "lua", "log", doc, logTo)
+	end
+end
+
+return logproxy
 \ No newline at end of file
 local skynet = require "skynet"
-local harbor = require "skynet.harbor"
+require "utils.init"
+
+local redisd_count = tonumber(skynet.getenv("thread"))
+local redisd
+skynet.init(function()
+	local idx = math.randomInt(1, redisd_count)
+	redisd = skynet.localname(".redis" .. idx)
+end)
 local table_insert = table.insert
 local redisproxy = {}
-
 local isUsePika = false