Lua 协程深度应用：协程调度器实现、异步任务编排与多协程并发控制

在这个每秒处理百万请求的互联网时代，我们的代码就像指挥交通的警察——需要同时处理多个任务而不堵塞道路。Lua协程就像这个场景里的超级交警，不需要复杂线程切换就能优雅完成多任务调度。今天让我们走进这个微小的虚拟机世界，看看协程如何在不同场景下绽放异彩。

一、协程基本功重练

1.1 协程的生命周期

Lua中的协程就像快递员小哥的工作流程：

-- 创建协程（技术栈：Lua 5.3）
local my_task = coroutine.create(function()
    print("已接单")
    coroutine.yield()  -- 送到快递柜暂存
    print("开始派送")
end)

print(coroutine.status(my_task))  --> suspended
coroutine.resume(my_task)        --> 打印"已接单"
print(coroutine.status(my_task))  --> suspended
coroutine.resume(my_task)        --> 打印"开始派送"
print(coroutine.status(my_task))  --> dead

1.2 yield与resume的默契配合

这对搭档就像在玩抛接球游戏：

local cooper = coroutine.create(function(a)
    print("收到"..a)
    local b = coroutine.yield("请回复")
    print("收到回复："..b)
    return "对话结束"
end)

local status, msg = coroutine.resume(cooper, "你好吗？")
print(msg) --> 请回复

status, msg = coroutine.resume(cooper, "我很好！")
print(msg) --> 对话结束

二、调度器设计的奥秘

2.1 生产者-消费者模式实战

让我们实现一个简易调度系统：

-- 任务调度器（技术栈：Lua 5.4）
local scheduler = {
    tasks = {},  -- 任务队列
    add_task = function(self, task)
        table.insert(self.tasks, coroutine.create(task))
    end,
    
    run = function(self)
        while #self.tasks > 0 do
            local task = table.remove(self.tasks, 1)
            local success = coroutine.resume(task)
            
            if coroutine.status(task) ~= "dead" then
                table.insert(self.tasks, task)
            end
        end
    end
}

-- 生产者任务
local function producer()
    for i=1,3 do
        print("生产零件"..i)
        coroutine.yield()
    end
end

-- 消费者任务
local function consumer()
    for i=1,2 do
        print("组装成品"..i)
        coroutine.yield()
    end
end

scheduler:add_task(producer)
scheduler:add_task(consumer)
scheduler:run()

/* 输出结果：
生产零件1
组装成品1
生产零件2
组装成品2
生产零件3
*/

2.2 事件驱动调度

模拟网络请求处理场景：

local event_loop = {
    handlers = {},
    register = function(self, event_type, handler)
        self.handlers[event_type] = self.handlers[event_type] or {}
        table.insert(self.handlers[event_type], coroutine.create(handler))
    end,
    
    trigger = function(self, event_type, data)
        local handlers = self.handlers[event_type] or {}
        for _, co in ipairs(handlers) do
            if coroutine.status(co) == "suspended" then
                coroutine.resume(co, data)
            end
        end
    end
}

-- 模拟HTTP请求
event_loop:register("http_request", function()
    print("开始处理请求...")
    coroutine.yield()  -- 等待IO操作
    print("收到响应，处理完成")
end)

event_loop:trigger("http_request") --> 开始处理请求...
-- 假设1秒后收到响应
os.execute("sleep 1")  -- 模拟网络延迟
event_loop:trigger("http_request") --> 收到响应，处理完成

三、异步任务编排的魔法

3.1 并行任务处理

使用协程实现类似Promise.all的功能：

local function async_all(...)
    local tasks = {...}
    local results = {}
    
    local wrapper = function(index, task)
        return function()
            results[index] = task()
        end
    end

    local scheduler = {
        run = function(self)
            local cos = {}
            for i, task in ipairs(tasks) do
                cos[i] = coroutine.create(wrapper(i, task))
            end
            
            while true do
                local all_done = true
                for _, co in ipairs(cos) do
                    if coroutine.status(co) ~= "dead" then
                        all_done = false
                        coroutine.resume(co)
                    end
                end
                if all_done then break end
            end
            
            return results
        end
    }
    
    return scheduler:run()
end

-- 测试并行任务
local result = async_all(
    function() return "数据加载完成" end,
    function() return "缓存更新完毕" end,
    function() return "日志写入成功" end
)

print(table.concat(result, " | ")) --> 数据加载完成 | 缓存更新完毕 | 日志写入成功

四、并发控制的精妙平衡

4.1 信号量实现资源控制

模拟数据库连接池场景：

local function create_semaphore(max)
    local count = 0
    return {
        acquire = function()
            while count >= max do
                coroutine.yield()
            end
            count = count + 1
        end,
        
        release = function()
            count = count - 1
        end
    }
end

local db_pool = create_semaphore(2)  -- 最大2个连接

local function query_db(query)
    db_pool:acquire()
    print("执行查询："..query)
    coroutine.yield()  -- 模拟查询耗时
    db_pool:release()
end

-- 创建5个查询任务
local scheduler = {
    tasks = {},
    add = function(self, task)
        table.insert(self.tasks, coroutine.create(task))
    end,
    
    run = function(self)
        while #self.tasks > 0 do
            local task = table.remove(self.tasks, 1)
            coroutine.resume(task)
            if coroutine.status(task) ~= "dead" then
                table.insert(self.tasks, task)
            end
        end
    end
}

for i=1,5 do
    scheduler:add(function() query_db("SELECT * FROM table"..i) end)
end

scheduler:run()

/* 输出示例：
执行查询：SELECT * FROM table1
执行查询：SELECT * FROM table2
执行查询：SELECT * FROM table3
执行查询：SELECT * FROM table4
执行查询：SELECT * FROM table5
*/

五、真实世界的应用场景

5.1 游戏逻辑处理

在游戏开发中，每个NPC的行为控制都可以看作一个协程：

local npc_ai = coroutine.create(function()
    while true do
        print("巡逻中...")
        coroutine.yield()
        
        print("发现玩家！")
        coroutine.yield()
        
        print("发动攻击！")
        coroutine.yield()
    end
end)

-- 每帧执行
for _=1,5 do
    coroutine.resume(npc_ai)
end

5.2 网络服务处理

模拟HTTP服务器路由：

local router = {
    routes = {},
    add_route = function(self, path, handler)
        self.routes[path] = coroutine.create(handler)
    end,
    
    handle_request = function(self, path)
        local co = self.routes[path]
        if co and coroutine.status(co) == "suspended" then
            coroutine.resume(co)
        end
    end
}

router:add_route("/api/data", function()
    coroutine.yield()  -- 等待数据库查询
    print("返回JSON数据")
end)

router:handle_request("/api/data") --> 返回JSON数据

六、技术优缺点分析

6.1 优势亮点

• 轻量级：每个协程仅需2KB内存（Lua 5.3+）
• 精确控制：能手动管理执行流程
• 无锁编程：避免多线程竞争问题

6.2 注意事项

• 避免阻塞操作（如长时间循环）
• 注意协程泄漏问题
• 合理设计yield点位置

七、结语与最佳实践

当我们在Lua协程的海洋里遨游时，记住这些导航法则：

1. 为I/O密集型操作设计专用调度器
2. 给长期运行的协程设置超时机制
3. 保持协程功能的单一性
4. 用装饰器模式封装常用逻辑