Skynet 性能调优实战

-- lualib/benchmark.lua
local skynet = require "skynet"

local Benchmark = {}

function Benchmark.run(func, iterations, concurrency)
    iterations = iterations or 1000
    concurrency = concurrency or 10
    
    local results = {
        total_time = 0,
        min_time = math.huge,
        max_time = 0,
        success = 0,
        failed = 0,
    }
    
    local start_time = skynet.now()
    
    -- 并发执行
    for i = 1, concurrency do
        skynet.fork(function()
            for j = 1, iterations / concurrency do
                local iter_start = skynet.now()
                local ok, err = pcall(func)
                local iter_time = skynet.now() - iter_start
                
                if ok then
                    results.success = results.success + 1
                    results.min_time = math.min(results.min_time, iter_time)
                    results.max_time = math.max(results.max_time, iter_time)
                else
                    results.failed = results.failed + 1
                end
            end
        end)
    end
    
    -- 等待所有协程完成
    skynet.sleep(iterations / concurrency / 100 + 10)
    
    results.total_time = skynet.now() - start_time
    results.avg_time = results.total_time / (results.success + results.failed)
    results.qps = (results.success + results.failed) / (results.total_time / 100)
    
    return results
end

function Benchmark.print_report(name, results)
    skynet.error(string.rep("=", 60))
    skynet.error(string.format("性能测试报告: %s", name))
    skynet.error(string.rep("=", 60))
    skynet.error(string.format("成功次数: %d", results.success))
    skynet.error(string.format("失败次数: %d", results.failed))
    skynet.error(string.format("总耗时: %.2f 秒", results.total_time / 100))
    skynet.error(string.format("平均耗时: %.2f ms", results.avg_time * 10))
    skynet.error(string.format("最小耗时: %.2f ms", results.min_time * 10))
    skynet.error(string.format("最大耗时: %.2f ms", results.max_time * 10))
    skynet.error(string.format("QPS: %.2f", results.qps))
    skynet.error(string.rep("=", 60))
end

return Benchmark

-- service/monolith.lua - 单体服务
local CMD = {}

function CMD.login(...) end
function CMD.chat(...) end
function CMD.battle(...) end
function CMD.payment(...) end
function CMD.mail(...) end

-- 问题：所有功能在一个服务中，无法水平扩展

-- 按功能拆分
-- service/login.lua - 登录服务
-- service/chat.lua - 聊天服务
-- service/battle.lua - 战斗服务（多实例）
-- service/payment.lua - 支付服务
-- service/mail.lua - 邮件服务

-- service/battle_dispatcher.lua
local skynet = require "skynet"

local battle_instances = {}
local instance_count = 5

skynet.start(function()
    -- 创建多个战斗服务实例
    for i = 1, instance_count do
        battle_instances[i] = skynet.newservice("battle", i)
    end
    
    skynet.dispatch("lua", function(session, source, cmd, ...)
        if cmd == "start_battle" then
            local player_id = ...
            -- 根据玩家 ID 分配到不同实例
            local instance_id = (player_id % instance_count) + 1
            local instance = battle_instances[instance_id]
            
            if session ~= 0 then
                skynet.retpack(skynet.call(instance, "lua", cmd, ...))
            else
                skynet.send(instance, "lua", cmd, ...)
            end
        end
    end)
end)

-- 不好：频繁发送小消息
for i = 1, 100 do
    skynet.send(target, "lua", "update_item", i, items[i])
end

-- 好：批量发送
skynet.send(target, "lua", "update_items", items)

-- service/data_aggregator.lua
local skynet = require "skynet"

local pending_updates = {}
local flush_timer = nil

local CMD = {}

function CMD.update(key, value)
    table.insert(pending_updates, {key = key, value = value})
    
    -- 达到阈值或定时器触发时批量写入
    if #pending_updates >= 100 then
        CMD.flush()
    elseif not flush_timer then
        flush_timer = skynet.timeout(100, function()
            CMD.flush()
            flush_timer = nil
        end)
    end
end

function CMD.flush()
    if #pending_updates == 0 then
        return
    end
    
    -- 批量写入数据库
    skynet.call(".db_service", "lua", "batch_update", pending_updates)
    pending_updates = {}
end

skynet.start(function()
    skynet.dispatch("lua", function(session, source, cmd, ...)
        local f = assert(CMD[cmd])
        if session ~= 0 then
            skynet.retpack(f(...))
        else
            f(...)
        end
    end)
end)

-- lualib/rate_limiter.lua
local skynet = require "skynet"

local RateLimiter = {}
RateLimiter.__index = RateLimiter

function RateLimiter.new(max_requests, time_window)
    return setmetatable({
        max_requests = max_requests,
        time_window = time_window,
        requests = {},
    }, RateLimiter)
end

function RateLimiter:allow(key)
    local now = skynet.time()
    
    -- 清理过期记录
    local valid_requests = {}
    for _, timestamp in ipairs(self.requests[key] or {}) do
        if now - timestamp < self.time_window then
            valid_requests[#valid_requests + 1] = timestamp
        end
    end
    
    -- 检查是否超过限制
    if #valid_requests >= self.max_requests then
        return false
    end
    
    -- 记录新请求
    valid_requests[#valid_requests + 1] = now
    self.requests[key] = valid_requests
    
    return true
end

-- 使用示例
local limiter = RateLimiter.new(100, 60)  -- 每分钟最多 100 次

function CMD.process_request(user_id, data)
    if not limiter:allow(user_id) then
        return false, "请求过于频繁"
    end
    
    -- 处理请求
    return do_process(data)
end

return RateLimiter

-- lualib/object_pool.lua
local skynet = require "skynet"

local ObjectPool = {}
ObjectPool.__index = ObjectPool

function ObjectPool.new(create_func, reset_func, initial_size)
    local pool = setmetatable({
        create = create_func,
        reset = reset_func,
        objects = {},
    }, ObjectPool)
    
    -- 预创建对象
    for i = 1, initial_size or 10 do
        pool.objects[#pool.objects + 1] = create_func()
    end
    
    return pool
end

function ObjectPool:acquire()
    if #self.objects > 0 then
        return table.remove(self.objects)
    end
    return self.create()
end

function ObjectPool:release(obj)
    if self.reset then
        self.reset(obj)
    end
    self.objects[#self.objects + 1] = obj
end

function ObjectPool:size()
    return #self.objects
end

return ObjectPool

-- 使用示例：子弹对象池
local BulletPool = ObjectPool.new(
    function()
        return {x = 0, y = 0, vx = 0, vy = 0, damage = 0}
    end,
    function(bullet)
        bullet.x = 0
        bullet.y = 0
        bullet.vx = 0
        bullet.vy = 0
        bullet.damage = 0
    end,
    100  -- 预创建 100 个
)

-- 获取子弹
local bullet = BulletPool:acquire()
bullet.x = 100
bullet.y = 200
bullet.damage = 50

-- 回收子弹
BulletPool:release(bullet)

-- lualib/lru_cache.lua
local skynet = require "skynet"

local LRUCache = {}
LRUCache.__index = LRUCache

function LRUCache.new(max_size, ttl)
    return setmetatable({
        max_size = max_size or 1000,
        ttl = ttl or 300,  -- 5 分钟
        cache = {},
        access_order = {},
    }, LRUCache)
end

function LRUCache:get(key)
    local entry = self.cache[key]
    if not entry then
        return nil
    end
    
    -- 检查过期
    if skynet.time() - entry.timestamp > self.ttl then
        self:remove(key)
        return nil
    end
    
    -- 更新访问顺序
    self:update_access(key)
    
    return entry.value
end

function LRUCache:set(key, value)
    -- 如果已存在，更新
    if self.cache[key] then
        self.cache[key].value = value
        self.cache[key].timestamp = skynet.time()
        self:update_access(key)
        return
    end
    
    -- 如果已满，删除最久未使用的
    if #self.access_order >= self.max_size then
        local oldest_key = table.remove(self.access_order, 1)
        self.cache[oldest_key] = nil
    end
    
    -- 添加新条目
    self.cache[key] = {
        value = value,
        timestamp = skynet.time(),
    }
    self.access_order[#self.access_order + 1] = key
end

function LRUCache:remove(key)
    self.cache[key] = nil
    for i, k in ipairs(self.access_order) do
        if k == key then
            table.remove(self.access_order, i)
            break
        end
    end
end

function LRUCache:update_access(key)
    for i, k in ipairs(self.access_order) do
        if k == key then
            table.remove(self.access_order, i)
            break
        end
    end
    self.access_order[#self.access_order + 1] = key
end

function LRUCache:clear()
    self.cache = {}
    self.access_order = {}
end

return LRUCache

-- 使用示例
local user_cache = LRUCache.new(10000, 600)  -- 10000 用户，10 分钟过期

function get_user(user_id)
    local user = user_cache:get(user_id)
    if user then
        return user
    end
    
    -- 从数据库加载
    user = skynet.call(".db_service", "lua", "get_user", user_id)
    if user then
        user_cache:set(user_id, user)
    end
    
    return user
end

-- 不好：使用字符串键（哈希表，开销大）
local config = {}
config["max_players"] = 100
config["timeout"] = 30
config["retry_count"] = 3

-- 好：使用数字索引（数组，开销小）
local config = {100, 30, 3}
local MAX_PLAYERS, TIMEOUT, RETRY_COUNT = 1, 2, 3

-- 更好：使用局部变量
local max_players = 100
local timeout = 30
local retry_count = 3

-- 不好：循环中拼接字符串
local result = ""
for i = 1, 10000 do
    result = result .. "x"  -- 每次创建新字符串
end

-- 好：使用 table.concat
local parts = {}
for i = 1, 10000 do
    parts[i] = "x"
end
local result = table.concat(parts)

-- 好：使用 string.rep
local result = string.rep("x", 10000)

-- 不好：所有请求都经过一个服务
local db_service = skynet.uniqueservice("db_service")

-- 好：按数据类型分片
local db_services = {}
for i = 1, 5 do
    db_services[i] = skynet.newservice("db_service", i)
end

function get_db_service(user_id)
    local shard = (user_id % 5) + 1
    return db_services[shard]
end

-- 不好：同步等待耗时操作
function CMD.process(data)
    local result = skynet.call(".slow_service", "lua", "compute", data)
    -- 阻塞当前协程
    return result
end

-- 好：异步处理，立即返回
function CMD.process(data)
    -- 立即返回任务 ID
    local task_id = generate_task_id()
    
    -- 后台处理
    skynet.fork(function()
        local result = skynet.call(".slow_service", "lua", "compute", data)
        -- 保存结果
        save_result(task_id, result)
        -- 通知客户端
        notify_client(task_id, result)
    end)
    
    return task_id
end

-- 客户端轮询结果
function CMD.get_result(task_id)
    return get_result(task_id)
end

-- service/data_service.lua
local skynet = require "skynet"

-- 读服务（多实例）
local read_services = {}
-- 写服务（单实例）
local write_service = nil

skynet.start(function()
    -- 创建 3 个读服务
    for i = 1, 3 do
        read_services[i] = skynet.newservice("data_reader", i)
    end
    
    -- 创建 1 个写服务
    write_service = skynet.newservice("data_writer")
    
    skynet.dispatch("lua", function(session, source, cmd, ...)
        if cmd == "read" then
            -- 轮询分发读请求
            local idx = (source % 3) + 1
            if session ~= 0 then
                skynet.retpack(skynet.call(read_services[idx], "lua", cmd, ...))
            else
                skynet.send(read_services[idx], "lua", cmd, ...)
            end
        elseif cmd == "write" then
            -- 写请求发送到写服务
            if session ~= 0 then
                skynet.retpack(skynet.call(write_service, "lua", cmd, ...))
            else
                skynet.send(write_service, "lua", cmd, ...)
            end
        end
    end)
end)

-- 不好：频繁访问全局变量
function process_data(data)
    for i = 1, #data do
        data[i] = math.sqrt(data[i])  -- math 是全局变量
    end
end

-- 好：缓存为局部变量
local sqrt = math.sqrt
local insert = table.insert
local concat = table.concat

function process_data(data)
    for i = 1, #data do
        data[i] = sqrt(data[i])
    end
end

-- 不好：重复计算
function update_positions(entities)
    for i = 1, #entities do
        local e = entities[i]
        e.x = e.x + e.vx * get_delta_time()  -- 每次都调用
        e.y = e.y + e.vy * get_delta_time()
    end
end

-- 好：提前计算
function update_positions(entities)
    local dt = get_delta_time()  -- 只计算一次
    for i = 1, #entities do
        local e = entities[i]
        e.x = e.x + e.vx * dt
        e.y = e.y + e.vy * dt
    end
end

-- 不好：O(n²) 查找
function find_entity(entities, target_id)
    for i = 1, #entities do
        if entities[i].id == target_id then
            return entities[i]
        end
    end
    return nil
end

-- 好：O(1) 查找（使用索引）
local entity_index = {}  -- id -> entity

function build_index(entities)
    for i = 1, #entities do
        entity_index[entities[i].id] = entities[i]
    end
end

function find_entity(target_id)
    return entity_index[target_id]
end

-- 不好：逐条插入
for i = 1, 1000 do
    skynet.call(".db", "lua", "insert", "users", users[i])
end

-- 好：批量插入
skynet.call(".db", "lua", "batch_insert", "users", users)

-- service/async_logger.lua
local skynet = require "skynet"

local log_buffer = {}
local buffer_size = 1000
local flush_interval = 100  -- 1 秒

local CMD = {}

function CMD.log(level, message)
    table.insert(log_buffer, {
        level = level,
        message = message,
        timestamp = os.time(),
    })
    
    -- 缓冲区满时立即刷新
    if #log_buffer >= buffer_size then
        CMD.flush()
    end
end

function CMD.flush()
    if #log_buffer == 0 then
        return
    end
    
    -- 批量写入文件
    local logs_to_write = log_buffer
    log_buffer = {}
    
    skynet.fork(function()
        write_to_file(logs_to_write)
    end)
end

skynet.start(function()
    -- 定时刷新
    skynet.fork(function()
        while true do
            skynet.sleep(flush_interval)
            CMD.flush()
        end
    end)
    
    skynet.dispatch("lua", function(session, source, cmd, ...)
        local f = assert(CMD[cmd])
        if session ~= 0 then
            skynet.retpack(f(...))
        else
            f(...)
        end
    end)
end)

-- config.production

-- 线程数（根据 CPU 核心数）
thread = 16

-- 消息队列大小
queue = 32

-- 启动服务
start = "main"

-- 日志级别（生产环境用 INFO）
loglevel = "info"

-- 关闭调试控制台（生产环境）
-- console = nil

-- 集群配置
cluster = "config.cluster"
node = "node1"

-- service/main.lua
local skynet = require "skynet"

skynet.start(function()
    skynet.error("服务器启动中...")
    
    -- 按依赖顺序启动服务
    -- 1. 基础服务
    skynet.uniqueservice("config_mgr")
    skynet.uniqueservice("db_service")
    skynet.uniqueservice("redis_service")
    
    -- 2. 业务服务
    skynet.uniqueservice("user_service")
    skynet.uniqueservice("game_service")
    
    -- 3. 网关服务（最后启动）
    local gate = skynet.newservice("gate")
    skynet.call(gate, "lua", "open", {
        port = 8888,
        maxclient = 10000,
    })
    
    skynet.error("服务器启动完成")
end)

-- service/monitor.lua
local skynet = require "skynet"

local thresholds = {
    cpu = 80,          -- CPU 使用率 80%
    memory = 85,       -- 内存使用率 85%
    queue = 1000,      -- 消息队列 1000
    latency = 100,     -- 延迟 100ms
}

local CMD = {}

function CMD.check()
    local launcher = skynet.localname(".launcher")
    local services = skynet.call(launcher, "lua", "list")
    
    local alerts = {}
    
    for addr in pairs(services) do
        local ok, stat = pcall(skynet.call, addr, "debug", "STAT")
        if ok then
            -- 检查消息队列
            if stat.queue > thresholds.queue then
                table.insert(alerts, {
                    type = "queue",
                    service = string.format(":%08x", addr),
                    value = stat.queue,
                    threshold = thresholds.queue,
                })
            end
            
            -- 检查内存
            local ok, mem = pcall(skynet.call, addr, "debug", "MEM")
            if ok and mem > 100 * 1024 * 1024 then  -- 100MB
                table.insert(alerts, {
                    type = "memory",
                    service = string.format(":%08x", addr),
                    value = mem,
                    threshold = 100 * 1024 * 1024,
                })
            end
        end
    end
    
    -- 发送告警
    if #alerts > 0 then
        skynet.call(".alert_service", "lua", "send_alerts", alerts)
    end
    
    return alerts
end

skynet.start(function()
    -- 每 10 秒检查一次
    skynet.fork(function()
        while true do
            skynet.sleep(1000)
            pcall(CMD.check)
        end
    end)
    
    skynet.dispatch("lua", function(session, source, cmd, ...)
        local f = assert(CMD[cmd])
        if session ~= 0 then
            skynet.retpack(f(...))
        else
            f(...)
        end
    end)
end)