利用OpenResty的ngx.semaphore实现轻量级协程同步，协调复杂并发任务

一、当协程“小伙伴”需要排队：为什么需要同步？

想象一下，你正在管理一个高效的快递分拣中心。这个中心有一群不知疲倦的“超级员工”，他们就是协程。与传统线程不同，这些协程员工非常“轻”，切换任务几乎不花时间，可以同时处理成千上万个包裹（请求）。OpenResty的环境，就是这样一个完美的协程分拣中心，它基于Nginx，但用Lua语言赋予了其非阻塞、高并发的超能力。

现在，场景来了：有一批特别重要的加急包裹（比如，需要调用多个外部API、进行复杂数据聚合的任务），必须按照特定顺序处理，或者同一时间只能允许有限几个员工进入特定区域（比如访问一个共享资源，防止过载）。如果让所有协程员工一拥而上，场面就会失控，可能出现数据错乱、资源争抢，甚至把下游服务“打挂”。

这时，我们就需要一个聪明的“调度员”，来协调这些动作飞快的协程员工。这个调度员不能太笨重，否则就浪费了协程的轻量优势；它必须足够高效，能融入OpenResty的事件驱动模型。OpenResty自带的 ngx.semaphore（信号量）模块，就是这样一个理想的轻量级同步工具。它本质上是一个计数器，用来控制协程的“等待”和“放行”，完美解决协程间的协调问题。

二、认识核心工具：ngx.semaphore 信号量

ngx.semaphore 可以理解为一个带有数字的通行证发放亭。这个数字代表了当前可用的通行证数量。

新建信号量：local semaphore = require "ngx.semaphore".new(初始数量)。这就好比建了一个亭子，里面初始就放了N张通行证。
获取通行证（等待）：semaphore:wait(超时时间)。一个协程员工走到亭子前，尝试取一张通行证。如果亭子里有（数量>0），他就直接拿走一张，然后继续干活；如果亭子里没了（数量=0），他就得在亭子边安静地排队等待，直到有新通行证放入，或者等到指定的超时时间。
归还通行证（释放）：semaphore:post(释放数量)。一个协程员工完成工作后，可以把他拿走的通行证还回亭子（通常还一张）。如果此时有别的协程在排队等待，亭子会立刻把这张证发给排队中的第一位，让他恢复工作。

这个过程是完全非阻塞的。协程在 wait 时排队，不会占用CPU；在 post 时唤醒其他协程，高效流转。下面，我们通过一个最简单的例子来感受一下。

技术栈：OpenResty + Lua

-- 示例1：基础信号量使用 - 模拟一个只有3个座位的休息室
local semaphore = require "ngx.semaphore"
local log = ngx.log
local ERR = ngx.ERR
local INFO = ngx.INFO

-- 创建一个初始有3个“座位”（资源）的信号量
local restroom_sem = semaphore.new(3)

-- 模拟5个协程都想使用休息室
for i = 1, 5 do
    ngx.thread.spawn(function()
        local co_id = tostring(ngx.thread.self()):sub(9) -- 获取协程ID简写，用于日志
        log(INFO, "协程[", co_id, "] 走到休息室门口，尝试获取座位...")

        -- 尝试获取一个座位（资源），最多等待2秒
        local ok, err = restroom_sem:wait(2)
        if not ok then
            log(ERR, "协程[", co_id, "] 等待超时，没能用上休息室")
            return
        end

        -- 成功进入休息室
        log(INFO, "协程[", co_id, "] 成功进入休息室，开始使用...")
        ngx.sleep(1) -- 模拟使用休息室1秒钟

        -- 使用完毕，离开并释放座位
        log(INFO, "协程[", co_id, "] 使用完毕，离开并释放座位")
        restroom_sem:post(1) -- 释放一个资源回信号量池
    end)
end

-- 主协程稍作等待，让子协程有足够时间运行
ngx.sleep(5)

代码解释：我们创建了一个初始值为3的信号量，模拟一个有3个座位的休息室。然后创建了5个协程（5个人）都想进去。前3个协程会立刻拿到“座位”进入。第4和第5个协程会在门口等待。当先进入的协程使用完毕（ngx.sleep(1)后），调用 post 释放座位，等待的协程中的一个便会立即获取并进入。如果等待超过2秒还没座位，则会超时返回错误。通过日志，你可以清晰地看到协程的排队和唤醒过程。

三、实战进阶：协调复杂并发任务

理解了基础操作，我们来看一个更贴近实际的场景：批量获取用户信息。假设我们需要从某个接口获取100个用户的详细信息，但该接口有频率限制，要求每秒最多调用10次。如果我们简单粗暴地启动100个协程去调用，势必会触发限流。此时，用信号量来控制“同时发起请求的协程数量”就再合适不过了。

技术栈：OpenResty + Lua

-- 示例2：使用信号量实现并发控制 - 遵守下游API的速率限制
local semaphore = require "ngx.semaphore"
local http = require "resty.http"
local cjson = require "cjson"
local log = ngx.log
local INFO = ngx.INFO
local WARN = ngx.WARN

-- 模拟的用户ID列表
local user_ids = {}
for i = 1, 100 do
    user_ids[i] = 10000 + i
end

-- 创建一个信号量，初始数量为0。这代表“可用的并发配额”。
-- 我们将用一个单独的“配额发放”协程来控制每秒发放10个配额。
local quota_sem = semaphore.new(0)

-- 启动配额发放协程
ngx.thread.spawn(function()
    while true do
        -- 每秒向信号量中放入10个“配额”，允许10个请求发起
        quota_sem:post(10)
        log(INFO, "[配额发放者] 新发放了10个请求配额")
        ngx.sleep(1) -- 等待1秒
        -- 在实际应用中，这里可能需要一个更优雅的退出机制
    end
end)

-- 用于保存所有获取用户信息的子协程
local threads = {}

-- 为每个用户ID创建一个处理协程
for _, user_id in ipairs(user_ids) do
    local co = ngx.thread.spawn(function()
        -- 1. 等待获取一个请求配额
        local ok, err = quota_sem:wait(5) -- 最多等5秒拿配额
        if not ok then
            log(WARN, "为用户[", user_id, "]获取请求配额超时，任务放弃")
            return nil, "timeout"
        end

        -- 2. 拿到配额，执行实际的HTTP请求
        log(INFO, "为用户[", user_id, "]获取到配额，开始请求API...")
        local httpc = http.new()
        local res, req_err = httpc:request_uri("http://mock-api/users/" .. user_id, {
            method = "GET",
            headers = { ["Content-Type"] = "application/json" }
        })

        -- 3. 处理请求结果
        if not res then
            log(WARN, "请求用户[", user_id, "]失败: ", req_err)
            return nil, req_err
        end

        if res.status == 200 then
            local user_data = cjson.decode(res.body)
            log(INFO, "成功获取用户[", user_id, "]数据: ", user_data.name)
            return { id = user_id, data = user_data }
        else
            log(WARN, "请求用户[", user_id, "]异常，状态码: ", res.status)
            return nil, "http_status_" .. res.status
        end
    end)
    threads[#threads + 1] = co
end

-- 等待所有子协程完成，并收集结果（此处为简化，仅作演示）
-- 在实际中，你可能需要遍历 threads 数组，使用 ngx.thread.wait 来收集结果。
log(INFO, "所有用户数据获取任务已分发完毕，正在受控执行中...")

代码解释：这个示例展示了如何用信号量做速率限制。我们创建了一个初始为0的信号量 quota_sem。一个独立的“配额发放”协程每秒执行一次 post(10)，即每秒增加10个可用配额。100个处理用户请求的协程，在发起HTTP请求前，都必须先调用 quota_sem:wait() 获取一个配额。这就保证了无论有多少个请求协程，同一时刻最多只有10个能真正进入请求阶段，完美遵守了下游API的限流规则。这是一种非常经典的“生产者-消费者”模型变体。

四、更多应用场景与深度剖析

信号量的用途远不止于此，它在协调复杂并发任务时非常灵活。

应用场景：

数据库连接池限流：控制同时访问数据库的协程数量，防止数据库连接耗尽。
批量任务分阶段同步：一个任务需要A、B两个阶段，所有协程完成A阶段后，才能一起进入B阶段。可以使用一个初始为0的信号量，在所有协程完成A后，由一个协调者 post 足够数量来唤醒所有协程进入B。
替代繁忙等待：在需要轮询某个状态直到条件满足时，用信号量让协程等待，由条件达成者 post 唤醒，比循环查询更高效。
实现简单的互斥锁：创建一个初始数量为1的信号量，就变成了一个互斥锁（Mutex），保证同一时间只有一个协程进入临界区。

技术优缺点：

优点：
- 极其轻量：作为用户态构造，比操作系统线程锁或进程间通信机制开销小得多。
- 无缝集成：专为OpenResty的Lua协程环境设计，与Nginx事件模型完美契合，不会阻塞Worker进程。
- 使用简单：API非常简洁，核心就 new、wait、post 三个操作。
- 灵活可控：可以方便地实现限流、同步、互斥等多种模式。
缺点：
- 作用域有限：信号量对象只在单个Nginx Worker进程内有效。跨进程的协程无法通过它同步。这意味着如果你的OpenResty配置了多个Worker，每个Worker有自己独立的信号量世界。
- 无持久化：信号量状态存在于内存中，服务重启或Worker异常退出即消失。
- 需要手动管理：需要开发者仔细设计 post 和 wait 的逻辑，否则容易导致死锁（协程永远等待）或资源泄漏（忘了归还配额）。

注意事项：

死锁预防：确保每个 wait 在超时或正常逻辑路径上，最终都有对应的 post 被调用。仔细检查异常处理分支。
超时设置：总是为 wait 操作设置一个合理的超时时间，这是避免协程永久挂死的重要防线。
跨Worker问题：如果同步需求涉及多个Worker进程，你需要借助外部存储如 Redis 或 共享内存字典（lua_shared_dict） 配合信号量来实现。通常模式是：用外部存储做状态判断，用本Worker内的信号量做协程调度。
初始化值：new(0) 常用于创建需要外部触发的同步点；new(N) 常用于创建资源池或并发控制器。

让我们看一个结合 lua_shared_dict 实现跨Worker初始化的例子：

技术栈：OpenResty + Lua

-- 示例3：结合共享内存，实现多Worker间的协同启动（仅首次初始化）
local semaphore = require "ngx.semaphore"
local shared_data = ngx.shared.my_shared_dict -- 假设在nginx.conf中已定义
local log = ngx.log
local INFO = ngx.INFO

-- 每个Worker在启动时都会执行这段代码
local init_sem = semaphore.new(0) -- 初始为0，所有Worker都等待

local key = "my_app_initialized"
-- 尝试在共享内存中设置一个标记，只有一个Worker会成功
local success, err, forcible = shared_data:add(key, true, 60) -- 60秒过期

if success then
    -- 这个Worker是执行初始化的“领导者”
    log(INFO, "Worker[", ngx.worker.pid(), "] 担任领导者，开始执行全局初始化任务...")
    ngx.sleep(2) -- 模拟耗时的初始化，如加载大型数据到共享内存
    log(INFO, "全局初始化完成！")
    -- 初始化完成后，唤醒本Worker内所有等待的协程
    -- 注意：这里需要知道有多少个协程在等待，或者直接post一个足够大的数。
    -- 这里为简单，我们假设只有主协程在等，所以post(1)
    init_sem:post(1)
else
    -- 其他Worker是“跟随者”，等待初始化完成
    log(INFO, "Worker[", ngx.worker.pid(), "] 等待全局初始化...")
    init_sem:wait(10) -- 等待领导者完成初始化，最多10秒
    log(INFO, "Worker[", ngx.worker.pid(), "] 收到通知，继续启动。")
end
-- 所有Worker此后都可以安全使用初始化好的共享资源了

代码解释：这个例子解决了多个Nginx Worker进程可能重复初始化共享资源的问题。我们利用共享内存的 add 操作的原子性，确保只有一个Worker（领导者）执行真正的初始化任务。其他Worker（跟随者）则在一个信号量上等待。领导者初始化完成后，调用 post 唤醒自己Worker内等待的协程。注意，这个信号量只能同步同一个Worker内部的协程。不同Worker的跟随者，实际上是在各自独立的信号量上等待。这种模式常用于服务启动时的资源加载。

五、总结

ngx.semaphore 是OpenResty协程世界里一把小巧而强大的瑞士军刀。它通过简单的计数操作，为高速并发的协程提供了可靠的“交通管制”能力。无论是控制并发度、实现阶段同步，还是构建简单的资源池，它都能出色地完成任务。

其核心价值在于“轻量”与“契合”，它生来就是为OpenResty的非阻塞架构服务的。当然，它的局限性（进程内有效）也要求我们在设计架构时心中有数，对于更复杂的分布式同步需求，需要将其与Redis、共享内存等外部组件结合起来使用。

掌握 ngx.semaphore，意味着你能够更精细地驾驭OpenResty的并发能力，写出既高效又稳健的高性能应用。下次当你的协程“小伙伴”们需要排队或握手时，不妨考虑请出这位低调的调度员。

利用OpenResty的ngx.semaphore实现轻量级协程同步，协调复杂并发任务

一、当协程“小伙伴”需要排队：为什么需要同步？

二、认识核心工具：ngx.semaphore 信号量

三、实战进阶：协调复杂并发任务

四、更多应用场景与深度剖析

五、总结

评论

关联文章

一、 当协程“小伙伴”需要排队：为什么需要同步？

二、 认识核心工具：ngx.semaphore 信号量

三、 实战进阶：协调复杂并发任务

四、 更多应用场景与深度剖析

五、 总结

评论

关联文章

一、当协程“小伙伴”需要排队：为什么需要同步？

二、认识核心工具：ngx.semaphore 信号量

三、实战进阶：协调复杂并发任务

四、更多应用场景与深度剖析

五、总结