openresty(lua-API-轻量级线程-ngx.thread)

距离上一次更新该文章已经过了 689 天，文章所描述的內容可能已经发生变化，请留意。

摘自：http://www.daileinote.com/computer/openresty/14

OpenResty ngx_lua 模块的 ngx.thread(轻量级线程) 可以实现并行的访问上游服务，其内部实现原理就是基于 Lua 的协程，只是由 ngx_lua 帮忙调度，隐藏了一些细节处理。

由 rewrite_by_lua、access_by_lua、content_by_lua 等运行的 Lua 代码本身就是协程，我们可以称之为入口点协程或者主协程，然后再由该协程调用 ngx.thread.spawn 再生成协程，可以称之为子协程或者用户协程。

默认情况下当前的handler(比如由 content_by_lua* 的内容阶段)是不会终止的，除非发生以下3中情况中的一种：

1.当主协程和所有用户协程都终止
2.当主协程或者子协程中的其中一个调用了 ngx.exit、ngx.exec、ngx.redirect、ngx.req.set_uri(uri, true)其中一种。
3.主协程发生错误终止

注意：其中一个子协程发生 Lua 错误并不会影响到其他子协程运行。

由于 nginx 自身的机制，一般情况下，在处理子请求时是不允许终止的。所以协程也一样，在处理子请求时也是不能终止，所以主协程必须要使用 ngx.thread.wait 来等待这些协程终止才能结束请求。不过调用 ngx.exit，传入状态码为 ngx.ERROR(-1),408,444,499 可以终止正在处理子请求的协程。

注意，协程是运行在非抢占式模式，所以它会独占cpu，除非

1.一次非阻塞io在一次运行不能完成
2.主动调用 coroutine.yield 交出执行权
3.发生 Lua 错误或者调用 ngx.exit、ngx.exec、ngx.redirect、ngx.req.set_uri(uri, true)其中一种。

前2中一般在稍后会重新运行，除非当前handler终止。

通过 coroutine.status() 可以获取协程状态，如果当前协程已经结束(不管是正常终止还是报错)，父协程还在运行但是却没有调用 ngx.thread.wait，那么该协程的状态为僵尸状态(zombie)。

lua

 local yield = coroutine.yield   --别名

 function f()
     local self = coroutine.running() -- 获取本协程
     ngx.say("f 1")
     yield(self)  --主动让出-1
     ngx.say("f 2")
     yield(self)  --主动让出-3
     ngx.say("f 3") --5
 end

 local self = coroutine.running()  --返回当前协程
 ngx.say("0")
 yield(self)  --主动让出cpu，但是目前只有一个协程，所以无效

 ngx.say("1")

 --生成另外一个协程并进入运行
 ngx.thread.spawn(f)  --0

 ngx.say("2")
 yield(self)  -- 主动让出-2

 ngx.say("3")
 yield(self)  --主动让出-4

 ngx.say("4") --6
0
1
f 1
2
f 2
3
f 3
4

ngx.thread.wait

plaintext

1	syntax: ok, res1, res2, ... = ngx.thread.wait(thread1, thread2, ...)

等待一个或多个子协程终止，thread1，thread2 都是由 ngx.thread.spawn 返回的。

只有直接父协程才有资格 wait 自己的子协程。

bash

content_by_lua_block {
    
    local capture = ngx.location.capture
    local spawn = ngx.thread.spawn
    local wait = ngx.thread.wait
    local say = ngx.say

    local function fetch(uri)
        return {status=1,body=uri}
    end

    local threads = {
        spawn(fetch, "/foo"),
        spawn(fetch, "/bar"),
        spawn(fetch, "/baz")
    }

    for i = 1, #threads do
        local ok, res = wait(threads[i])
        if not ok then
            say(i, ": failed to run: ", res)
        else
            say(i, ": status: ", res.status)
            say(i, ": body: ", res.body)
        end
    end
    
}
1: status: 1
1: body: /foo
2: status: 1
2: body: /bar
3: status: 1
3: body: /baz