请帮review代码 stream_lua_module tcp server 处理消息写数据 ngx.thread.spawn用法姿势是否正确？

winlia520

client端代码
--
-- Created by IntelliJ IDEA.
-- User: winfan
-- Date: 16/12/8
-- Time: 上午8:56

local cjson = require "cjson.safe"
local logger = require "resty.logger.socket"
local common = require "atomic.common"
local config = require "atomic.config"
local uri = ngx.var.uri
local get_req = ngx.req.get_uri_args()
local req_info = {}
if uri == '/xx' or uri == '/xx/' then
    if not logger.initted() then
        local ok, err = logger.init(config.collect_log_service_1)
        if not ok then
            ngx.log(ngx.ERR, "failed to initialize the logger: ", err)
            return
        end
    end
    req_info['policy_id'] = get_req["policy_id"]
    req_info['number'] = get_req["number"]
    req_info['version'] = get_req["version"]
    req_info["ip"] = common.get_client_ip()
    local info = {
        type = "collect_num",
        data = ""
    }
    local logstr = cjson.encode(info)
    local bytes, err = logger.log(logstr).."\n"  --这里加个换行，防粘包处理
    if err then
        ngx.log(ngx.ERR, "failed to log message: ", err)
        return
    end

end


server端代码

local config = require "atomic.config"
local hand_mysql = require "atomic.hand_mysql"
local common = require "atomic.common"
local cjson = require "cjson.safe"
local logdao = require "atomic.log"
local sock = assert(ngx.req.socket(true))
sock:settimeout(1000)  -- one second timout
local data = sock:receive('*l')
if not common.is_empty(data) then
    ngx.log(ngx.ALERT,"服务端接收数据：",data)
    --我这里开了10个轻线程处理消息，对处理结果并不关心，没用到wait
    for i = 1, 10 do
        ngx.thread.spawn(
           function ()
               --ngx.log(ngx.ERR,"服务端接收数据：",data)
               local req_log_info = common.url_decode(data)
               if req_log_info == nil then
                   ngx.log(ngx.ERR,"服务端接收数据解析错误：",data)
                   return 0
               end
               --连接数据连接池
               local db = hand_mysql:new(config.log_db_conf)
               --把日志写到数据库
               logdao.save_upgrade_download_log(db, req_log_info['data'])
           end
        )
    end
    return 1
else
   ngx.log(ngx.ERR,"服务端接收数据：",data)
end

winlia520

server端代码发现问题，一条消息会插入10次！！！

应该怎样实现一条消息只分配给一个线程处理，把接收消息也放到for循环里面吗？

在 2017年3月7日星期二 UTC+8下午2:44:49，tan winfan写道：

client端代码
--
-- Created by IntelliJ IDEA.
-- User: winfan
-- Date: 16/12/8
-- Time: 上午8:56

local cjson = require "cjson.safe"
local logger = require "resty.logger.socket"
local common = require "atomic.common"
local config = require "atomic.config"
local uri = ngx.var.uri
local get_req = ngx.req.get_uri_args()
local req_info = {}
if uri == '/xx' or uri == '/xx/' then
    if not logger.initted() then
        local ok, err = logger.init(config.collect_log_service_1)
        if not ok then
            ngx.log(ngx.ERR, "failed to initialize the logger: ", err)
            return
        end
    end
    req_info['policy_id'] = get_req["policy_id"]
    req_info['number'] = get_req["number"]
    req_info['version'] = get_req["version"]
    req_info["ip"] = common.get_client_ip()
    local info = {
        type = "collect_num",
        data = ""
    }
    local logstr = cjson.encode(info)
    local bytes, err = logger.log(logstr).."\n"  --这里加个换行，防粘包处理
    if err then
        ngx.log(ngx.ERR, "failed to log message: ", err)
        return
    end

end


server端代码

local config = require "atomic.config"
local hand_mysql = require "atomic.hand_mysql"
local common = require "atomic.common"
local cjson = require "cjson.safe"
local logdao = require "atomic.log"
local sock = assert(ngx.req.socket(true))
sock:settimeout(1000)  -- one second timout
local data = sock:receive('*l')
if not common.is_empty(data) then
    ngx.log(ngx.ALERT,"服务端接收数据：",data)
    --我这里开了10个轻线程处理消息，对处理结果并不关心，没用到wait
    for i = 1, 10 do
        ngx.thread.spawn(
           function ()
               --ngx.log(ngx.ERR,"服务端接收数据：",data)
               local req_log_info = common.url_decode(data)
               if req_log_info == nil then
                   ngx.log(ngx.ERR,"服务端接收数据解析错误：",data)
                   return 0
               end
               --连接数据连接池
               local db = hand_mysql:new(config.log_db_conf)
               --把日志写到数据库
               logdao.save_upgrade_download_log(db, req_log_info['data'])
           end
        )
    end
    return 1
else
   ngx.log(ngx.ERR,"服务端接收数据：",data)
end

winlia520

local config = require "atomic.config"
local hand_mysql = require "atomic.hand_mysql"
local common = require "atomic.common"
local cjson = require "cjson.safe"
local logdao = require "atomic.log"
local sock = assert(ngx.req.socket(true))
sock:settimeout(1000) -- one second timout

for i = 1, 10 do
    ngx.thread.spawn(function()

        local data = "">
        if not common.is_empty(data) then
            ngx.log(ngx.ALERT, "服务端接收数据：", data) 
            local req_log_info = common.url_decode(data)
            if req_log_info == nil then
                ngx.log(ngx.ERR, "服务端接收数据解析错误：", data)
                return 0
            end
            --连接数据连接池
            local db = hand_mysql:new(config.log_db_conf)
            --把日志写到数据库
            logdao.save_upgrade_download_log(db, req_log_info['data'])
            return 1
        else
            ngx.log(ngx.ERR, "服务端接收数据：", data)
        end
    end)
end

server端改成以上这样？

在 2017年3月7日星期二 UTC+8下午2:44:49，tan winfan写道：

client端代码
--
-- Created by IntelliJ IDEA.
-- User: winfan
-- Date: 16/12/8
-- Time: 上午8:56

local cjson = require "cjson.safe"
local logger = require "resty.logger.socket"
local common = require "atomic.common"
local config = require "atomic.config"
local uri = ngx.var.uri
local get_req = ngx.req.get_uri_args()
local req_info = {}
if uri == '/xx' or uri == '/xx/' then
    if not logger.initted() then
        local ok, err = logger.init(config.collect_log_service_1)
        if not ok then
            ngx.log(ngx.ERR, "failed to initialize the logger: ", err)
            return
        end
    end
    req_info['policy_id'] = get_req["policy_id"]
    req_info['number'] = get_req["number"]
    req_info['version'] = get_req["version"]
    req_info["ip"] = common.get_client_ip()
    local info = {
        type = "collect_num",
        data = ""
    }
    local logstr = cjson.encode(info)
    local bytes, err = logger.log(logstr).."\n"  --这里加个换行，防粘包处理
    if err then
        ngx.log(ngx.ERR, "failed to log message: ", err)
        return
    end

end


server端代码

local config = require "atomic.config"
local hand_mysql = require "atomic.hand_mysql"
local common = require "atomic.common"
local cjson = require "cjson.safe"
local logdao = require "atomic.log"
local sock = assert(ngx.req.socket(true))
sock:settimeout(1000)  -- one second timout
local data = sock:receive('*l')
if not common.is_empty(data) then
    ngx.log(ngx.ALERT,"服务端接收数据：",data)
    --我这里开了10个轻线程处理消息，对处理结果并不关心，没用到wait
    for i = 1, 10 do
        ngx.thread.spawn(
           function ()
               --ngx.log(ngx.ERR,"服务端接收数据：",data)
               local req_log_info = common.url_decode(data)
               if req_log_info == nil then
                   ngx.log(ngx.ERR,"服务端接收数据解析错误：",data)
                   return 0
               end
               --连接数据连接池
               local db = hand_mysql:new(config.log_db_conf)
               --把日志写到数据库
               logdao.save_upgrade_download_log(db, req_log_info['data'])
           end
        )
    end
    return 1
else
   ngx.log(ngx.ERR,"服务端接收数据：",data)
end

winlia520

压测发现以下日志错误

*2769 kevent() reported that connect() failed while handling client connection, client: 127.0.0.1, server: 0.0.0.0:6678

2017/03/07 16:25:27 [error] 23600#0: *2769 stream [lua] log.lua:33: save_upgrade_download_log(): ===err===:failed to receive packet header: closed while handling client connection, client: 127.0.0.1, server: 0.0.0.0:6678

2017/03/07 16:25:27 [error] 23600#0: *2769 kevent() reported that connect() failed while handling client connection, client: 127.0.0.1, server: 0.0.0.0:6678

在 2017年3月7日星期二 UTC+8下午2:44:49，tan winfan写道：

client端代码
--
-- Created by IntelliJ IDEA.
-- User: winfan
-- Date: 16/12/8
-- Time: 上午8:56

local cjson = require "cjson.safe"
local logger = require "resty.logger.socket"
local common = require "atomic.common"
local config = require "atomic.config"
local uri = ngx.var.uri
local get_req = ngx.req.get_uri_args()
local req_info = {}
if uri == '/xx' or uri == '/xx/' then
    if not logger.initted() then
        local ok, err = logger.init(config.collect_log_service_1)
        if not ok then
            ngx.log(ngx.ERR, "failed to initialize the logger: ", err)
            return
        end
    end
    req_info['policy_id'] = get_req["policy_id"]
    req_info['number'] = get_req["number"]
    req_info['version'] = get_req["version"]
    req_info["ip"] = common.get_client_ip()
    local info = {
        type = "collect_num",
        data = ""
    }
    local logstr = cjson.encode(info)
    local bytes, err = logger.log(logstr).."\n"  --这里加个换行，防粘包处理
    if err then
        ngx.log(ngx.ERR, "failed to log message: ", err)
        return
    end

end


server端代码

local config = require "atomic.config"
local hand_mysql = require "atomic.hand_mysql"
local common = require "atomic.common"
local cjson = require "cjson.safe"
local logdao = require "atomic.log"
local sock = assert(ngx.req.socket(true))
sock:settimeout(1000)  -- one second timout
local data = sock:receive('*l')
if not common.is_empty(data) then
    ngx.log(ngx.ALERT,"服务端接收数据：",data)
    --我这里开了10个轻线程处理消息，对处理结果并不关心，没用到wait
    for i = 1, 10 do
        ngx.thread.spawn(
           function ()
               --ngx.log(ngx.ERR,"服务端接收数据：",data)
               local req_log_info = common.url_decode(data)
               if req_log_info == nil then
                   ngx.log(ngx.ERR,"服务端接收数据解析错误：",data)
                   return 0
               end
               --连接数据连接池
               local db = hand_mysql:new(config.log_db_conf)
               --把日志写到数据库
               logdao.save_upgrade_download_log(db, req_log_info['data'])
           end
        )
    end
    return 1
else
   ngx.log(ngx.ERR,"服务端接收数据：",data)
end