如何实现在body_filter阶段进行异步处理

[yuqiquan]

Description

比如说：
body_filetr阶段主要处理步骤

1. 获取上游服务返回的请求头里的内容，将这些内容封装成一个json数据
2. 将这个数据上传到某一个服务，同时客户端接收到响应
   就是想要将步骤二进行异步处理，数据上传服务的这一个操作不影响客户端的响应

我起初的想法是在body_filter阶段创建一个线程，在线程中执行上报服务的操作，但是调用ngx.thread.spawn函数 发现
该阶段无法使用，然后我尝试使用ngx.timer.at(0,function) ,测试发现当function执行时间很长的时候，客户端发送第一次请求是正常的，第二次请求时，function执行还未结束，就会进行阻塞状态

这是我写的demo插件

-- test.lua
local ngx = ngx
local commonutil = require("apisix.plugins.util.commonutil")
local core = require("apisix.core")
local process = require("ngx.process")
local thread_spawn = ngx.thread.spawn
local isExcute = false
local schema = {
    type = "object",
    properties = {
    }
}

local plugin_name = "test"

local _M = {
    version = 0.1,
    priority = 70,
    name = plugin_name,
    schema = schema,
}

local isExcute = false

function _M.check_schema(conf, schema_type)
    return core.schema.check(schema, conf)
end
local function request()
    core.log.warn("发送上报请求 ")
    local num = 100000000000
    while num > 0 do
        num = num - 1
    end
end
--local function createThread ()
--    core.log.warn("创建线程进行监控")
--    local th, err = thread_spawn(request)
--    if not th then
--        core.log.error("failed to spawn thread : ", err)
--    end
--end

function _M.init()

end

function _M.destroy()
end

function _M.rewrite(conf, ctx)
end


function _M.access(conf, ctx)
end



function _M.header_filter(conf, ctx)
    core.response.set_header("Content-Type", "application/json;charset=utf-8")
    core.response.clear_header_as_body_modified()
end




function _M.body_filter(conf, ctx)
    local hearders = ngx.resp.get_headers()
    core.log.warn("response header --->  ", core.json.encode(hearders))
    -- 定义全局变量，收集全部响应
    if ngx.ctx.buffered == nil then
        ngx.ctx.buffered = {}
    end
    -- 如果非最后一次响应，将当前响应赋值
    if ngx.arg[1] ~= "" and not ngx.is_subrequest then
        table.insert(ngx.ctx.buffered, ngx.arg[1])
        -- 将当前响应赋值为空，以修改后的内容作为最终响应
        ngx.arg[1] = nil
    end
    local result = {}
    result.errcode = "200"
    result.errMes = "请求成功"
    if ngx.arg[2] == true then
        -- 获取所有响应数据
        local whole = table.concat(ngx.ctx.buffered)
        ngx.ctx.buffered = nil
        -- 进行你所需要进行的处理
        whole = core.json.encode(result)
        -- 重新赋值响应数据，以修改后的内容作为最终响应
        ngx.arg[1] = whole
        ngx.timer.at(0,request)
    end

end

function _M.delayed_body_filter(conf, ctx)

end

return _M

Environment

• APISIX version (run apisix version): 2.15.0

[tzssangglass]

function执行还未结束，就会进行阻塞状态

Can you be sure it's caused by ngx.timer.at(0,request)?
How about if you remove ngx.timer.at(0,request)?

[yuqiquan]

是的，删除ngx.timer.at(0,request)后，请求就正常了

[yuqiquan]

上面的demo例子，第二次发送请求时就会阻塞，一直到第一次请求发生的request函数执行完成，才能请求成功

[yuqiquan]

经过我的测试发现，ngx.timer.at(delay,function) , function执行时间 确实会对请求有影响

[tzssangglass]

ngx.timer.at does take CPU resources away from processing normal requests.

try with use ngx.sleep(0) to giving up CPU resources, refer to: groups.google.com/g/openresty/c/fuY_vTS01eg

example: knyar/nginx-lua-prometheus@ca178f4#diff-db8217cd3dee975eb8f254726ff765ee0d403bb8b5ea44994b0d58825324c5d9R821-R839