init_worker中通过ngx.timer.at写数据库的问题

xieqin1

Hi，我在init_worker中监听mqtt，收到数据则通过ngx.timer.at写数据库（postgresql），目前监听成功，但是ngx.timer.at总调用不到，不知各位有何建议？代码如下：

local function wt_nodedata(premature, data)

ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"]) ---------------------------2 监听到消息时这里无打印日志了

local pg = pgmoon.new({

host = "127.0.0.1",

port = "5432",

database = "postgre1",

user = "admin",

password = "dbpassword"

})

pg:connect()

local res = pg:query("“”)

pg:keepalive()

end

function callback(topic, payload)

ngx.log(ngx.ERR, "a data reveived topic:", topic)

--write to db

local data = "">

ngx.log(ngx.ERR, "a data reveived data:", data["devEUI"])

local ok, err = ngx.timer.at(0, wt_nodedata, data)

ngx.log(ngx.ERR, "1 create timer: ", ok) -----------------------------------1 监听到消息时这里还可以看到打印日志

ngx.log(ngx.ERR, "2 create timer: ", err)

if not ok then

ngx.log(ngx.ERR, "failed to create timer: ", err)

return

end

local MQTT = require "mqtt"

MQTT.Utility.set_debug(true)

MQTT.client.KEEP_ALIVE_TIME=60

local mqtt_client = MQTT.client.create('127.0.0.1', 1883, callback)

local error_message = mqtt_client:connect(tostring(ngx.worker.pid()))

local topics = {"application/+/rx"}

mqtt_client:subscribe(topics)

while (error_message == nil) do

error_message = mqtt_client:handler() -----------------------------------0 监听到消息时这里会调用callback函数

socket.sleep(1.0)

end

doujiang24

Hello，

这样用看起来是没有问题的，但是我不确定你的 mqtt 是怎么个实现

你可以先把 mqtt 部分代码去掉，先简化一下，直接执行你那个 callback，看是否可以

如果是 mqtt 部分的问题，那或许就得研究一下代码了 :)

2016-08-22 19:06 GMT+08:00 <xie...@gmail.com>:

Hi，我在init_worker中监听mqtt，收到数据则通过ngx.timer.at写数据库（postgresql），目前监听成功，但是ngx.timer.at总调用不到，不知各位有何建议？代码如下：

local function wt_nodedata(premature, data)
ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"]) ---------------------------2 监听到消息时这里无打印日志了
local pg = pgmoon.new({
host = "127.0.0.1",
port = "5432",
database = "postgre1",
user = "admin",
password = "dbpassword"
})

pg:connect()

local res = pg:query("“”)

pg:keepalive()
end

function callback(topic, payload)
ngx.log(ngx.ERR, "a data reveived topic:", topic)
--write to db
local data = "">
ngx.log(ngx.ERR, "a data reveived data:", data["devEUI"])

local ok, err = ngx.timer.at(0, wt_nodedata, data)
ngx.log(ngx.ERR, "1 create timer: ", ok) -----------------------------------1 监听到消息时这里还可以看到打印日志
ngx.log(ngx.ERR, "2 create timer: ", err)
if not ok then
ngx.log(ngx.ERR, "failed to create timer: ", err)
return
end
end

local MQTT = require "mqtt"
MQTT.Utility.set_debug(true)
MQTT.client.KEEP_ALIVE_TIME=60
local mqtt_client = MQTT.client.create('127.0.0.1', 1883, callback)
local error_message = mqtt_client:connect(tostring(ngx.worker.pid()))

local topics = {"application/+/rx"}
mqtt_client:subscribe(topics)

while (error_message == nil) do
error_message = mqtt_client:handler() -----------------------------------0 监听到消息时这里会调用callback函数
socket.sleep(1.0)
end

--

xieqin1

我直接这样调用：

local ok, err = ngx.timer.at(0, wt_nodedata, test)

wt_nodedata没有执行（里面的log没有打印）

在 2016年8月22日星期一 UTC+8下午11:12:27，doujiang写道：

Hello，

这样用看起来是没有问题的，但是我不确定你的 mqtt 是怎么个实现

你可以先把 mqtt 部分代码去掉，先简化一下，直接执行你那个 callback，看是否可以

如果是 mqtt 部分的问题，那或许就得研究一下代码了 :)

2016-08-22 19:06 GMT+08:00 <xie...@gmail.com>:
Hi，我在init_worker中监听mqtt，收到数据则通过ngx.timer.at写数据库（postgresql），目前监听成功，但是ngx.timer.at总调用不到，不知各位有何建议？代码如下：

local function wt_nodedata(premature, data)
ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"]) ---------------------------2 监听到消息时这里无打印日志了
local pg = pgmoon.new({
host = "127.0.0.1",
port = "5432",
database = "postgre1",
user = "admin",
password = "dbpassword"
})

pg:connect()

local res = pg:query("“”)

pg:keepalive()
end

function callback(topic, payload)
ngx.log(ngx.ERR, "a data reveived topic:", topic)
--write to db
local data = "">
ngx.log(ngx.ERR, "a data reveived data:", data["devEUI"])

local ok, err = ngx.timer.at(0, wt_nodedata, data)
ngx.log(ngx.ERR, "1 create timer: ", ok) -----------------------------------1 监听到消息时这里还可以看到打印日志
ngx.log(ngx.ERR, "2 create timer: ", err)
if not ok then
ngx.log(ngx.ERR, "failed to create timer: ", err)
return
end
end

local MQTT = require "mqtt"
MQTT.Utility.set_debug(true)
MQTT.client.KEEP_ALIVE_TIME=60
local mqtt_client = MQTT.client.create('127.0.0.1', 1883, callback)
local error_message = mqtt_client:connect(tostring(ngx.worker.pid()))

local topics = {"application/+/rx"}
mqtt_client:subscribe(topics)

while (error_message == nil) do
error_message = mqtt_client:handler() -----------------------------------0 监听到消息时这里会调用callback函数
socket.sleep(1.0)
end

--

moonbingbing

在 2016年8月23日上午9:18， <xie...@gmail.com>写道：

我直接这样调用：
local ok, err = ngx.timer.at(0, wt_nodedata, test)
wt_nodedata没有执行（里面的log没有打印）

只有直接调用都没有执行的话，我怀疑是你打印日志的那行代码出错了，比如data["mac"]不是字符串。你看看 nginx 的 error.log 里面有出错信息吗？

在 2016年8月22日星期一 UTC+8下午11:12:27，doujiang写道：
Hello，

这样用看起来是没有问题的，但是我不确定你的 mqtt 是怎么个实现

你可以先把 mqtt 部分代码去掉，先简化一下，直接执行你那个 callback，看是否可以

如果是 mqtt 部分的问题，那或许就得研究一下代码了 :)

2016-08-22 19:06 GMT+08:00 <xie...@gmail.com>:
Hi，我在init_worker中监听mqtt，收到数据则通过ngx.timer.at写数据库（postgresql），目前监听成功，但是ngx.timer.at总调用不到，不知各位有何建议？代码如下：

local function wt_nodedata(premature, data)
ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"]) ---------------------------2 监听到消息时这里无打印日志了
local pg = pgmoon.new({
host = "127.0.0.1",
port = "5432",
database = "postgre1",
user = "admin",
password = "dbpassword"
})

pg:connect()

local res = pg:query("“”)

pg:keepalive()
end

function callback(topic, payload)
ngx.log(ngx.ERR, "a data reveived topic:", topic)
--write to db
local data = "">
ngx.log(ngx.ERR, "a data reveived data:", data["devEUI"])

local ok, err = ngx.timer.at(0, wt_nodedata, data)
ngx.log(ngx.ERR, "1 create timer: ", ok) -----------------------------------1 监听到消息时这里还可以看到打印日志
ngx.log(ngx.ERR, "2 create timer: ", err)
if not ok then
ngx.log(ngx.ERR, "failed to create timer: ", err)
return
end
end

local MQTT = require "mqtt"
MQTT.Utility.set_debug(true)
MQTT.client.KEEP_ALIVE_TIME=60
local mqtt_client = MQTT.client.create('127.0.0.1', 1883, callback)
local error_message = mqtt_client:connect(tostring(ngx.worker.pid()))

local topics = {"application/+/rx"}
mqtt_client:subscribe(topics)

while (error_message == nil) do
error_message = mqtt_client:handler() -----------------------------------0 监听到消息时这里会调用callback函数
socket.sleep(1.0)
end

--

--

xieqin1

你好，我把代码里面关于mqtt的删除掉，wt_nodedata里面的log语句正常打印出来了：

local function wt_nodedata(premature, data)

ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"])

db_query.write_nodedata(data)

end

local test = {}

local ok, err = ngx.timer.at(0, wt_nodedata, test)

我现在在想，init_worker里面是不是只能有一个tcp连接？
在 2016年8月23日星期二 UTC+8上午10:02:25，WenMing写道：

在 2016年8月23日上午9:18， <xie...@gmail.com>写道：
我直接这样调用：
local ok, err = ngx.timer.at(0, wt_nodedata, test)
wt_nodedata没有执行（里面的log没有打印）

只有直接调用都没有执行的话，我怀疑是你打印日志的那行代码出错了，比如data["mac"]不是字符串。你看看 nginx 的 error.log 里面有出错信息吗？

在 2016年8月22日星期一 UTC+8下午11:12:27，doujiang写道：
Hello，

这样用看起来是没有问题的，但是我不确定你的 mqtt 是怎么个实现

你可以先把 mqtt 部分代码去掉，先简化一下，直接执行你那个 callback，看是否可以

如果是 mqtt 部分的问题，那或许就得研究一下代码了 :)

2016-08-22 19:06 GMT+08:00 <xie...@gmail.com>:
Hi，我在init_worker中监听mqtt，收到数据则通过ngx.timer.at写数据库（postgresql），目前监听成功，但是ngx.timer.at总调用不到，不知各位有何建议？代码如下：

local function wt_nodedata(premature, data)
ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"]) ---------------------------2 监听到消息时这里无打印日志了
local pg = pgmoon.new({
host = "127.0.0.1",
port = "5432",
database = "postgre1",
user = "admin",
password = "dbpassword"
})

pg:connect()

local res = pg:query("“”)

pg:keepalive()
end

function callback(topic, payload)
ngx.log(ngx.ERR, "a data reveived topic:", topic)
--write to db
local data = "">
ngx.log(ngx.ERR, "a data reveived data:", data["devEUI"])

local ok, err = ngx.timer.at(0, wt_nodedata, data)
ngx.log(ngx.ERR, "1 create timer: ", ok) -----------------------------------1 监听到消息时这里还可以看到打印日志
ngx.log(ngx.ERR, "2 create timer: ", err)
if not ok then
ngx.log(ngx.ERR, "failed to create timer: ", err)
return
end
end

local MQTT = require "mqtt"
MQTT.Utility.set_debug(true)
MQTT.client.KEEP_ALIVE_TIME=60
local mqtt_client = MQTT.client.create('127.0.0.1', 1883, callback)
local error_message = mqtt_client:connect(tostring(ngx.worker.pid()))

local topics = {"application/+/rx"}
mqtt_client:subscribe(topics)

while (error_message == nil) do
error_message = mqtt_client:handler() -----------------------------------0 监听到消息时这里会调用callback函数
socket.sleep(1.0)
end

--

--

xieqin1

按理说，timer会起一个新线程用cosocket去连接postgresql，应该可以的呀。

对了mqtt走的是luasocket。

在 2016年8月22日星期一 UTC+8下午7:06:11，xie...@gmail.com写道：

Hi，我在init_worker中监听mqtt，收到数据则通过ngx.timer.at写数据库（postgresql），目前监听成功，但是ngx.timer.at总调用不到，不知各位有何建议？代码如下：

local function wt_nodedata(premature, data)
ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"]) ---------------------------2 监听到消息时这里无打印日志了
local pg = pgmoon.new({
host = "127.0.0.1",
port = "5432",
database = "postgre1",
user = "admin",
password = "dbpassword"
})

pg:connect()

local res = pg:query("“”)

pg:keepalive()
end

function callback(topic, payload)
ngx.log(ngx.ERR, "a data reveived topic:", topic)
--write to db
local data = "">
ngx.log(ngx.ERR, "a data reveived data:", data["devEUI"])

local ok, err = ngx.timer.at(0, wt_nodedata, data)
ngx.log(ngx.ERR, "1 create timer: ", ok) -----------------------------------1 监听到消息时这里还可以看到打印日志
ngx.log(ngx.ERR, "2 create timer: ", err)
if not ok then
ngx.log(ngx.ERR, "failed to create timer: ", err)
return
end
end

local MQTT = require "mqtt"
MQTT.Utility.set_debug(true)
MQTT.client.KEEP_ALIVE_TIME=60
local mqtt_client = MQTT.client.create('127.0.0.1', 1883, callback)
local error_message = mqtt_client:connect(tostring(ngx.worker.pid()))

local topics = {"application/+/rx"}
mqtt_client:subscribe(topics)

while (error_message == nil) do
error_message = mqtt_client:handler() -----------------------------------0 监听到消息时这里会调用callback函数
socket.sleep(1.0)
end

yuansheng

刚刚看了一下你用的 lua 库，是不是这个？ https://github.com/geekscape/mqtt_lua/blob/master/lua/mqtt_library.lua

这个库是阻塞运行的。

openresty 自身的异步世界，是基于 cosocket 的，与 luasocket 区别很大。

luasocket 是阻塞执行，他会破坏 nginx 的异步世界，让所有的行为 stop。

这也就合理解释了 ngx.timer.at 没有执行的原因，因为 openresty 被 mqtt 阻塞了。

建议方法：

对 https://github.com/geekscape/mqtt_lua/blob/master/lua/mqtt_library.lua 动手术，修改成 cosocket 版本。

On Mon, Aug 22, 2016 at 7:06 PM, <xie...@gmail.com> wrote:

Hi，我在init_worker中监听mqtt，收到数据则通过ngx.timer.at写数据库（postgresql），目前监听成功，但是ngx.timer.at总调用不到，不知各位有何建议？代码如下：

local function wt_nodedata(premature, data)
ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"]) ---------------------------2 监听到消息时这里无打印日志了
local pg = pgmoon.new({
host = "127.0.0.1",
port = "5432",
database = "postgre1",
user = "admin",
password = "dbpassword"
})

pg:connect()

local res = pg:query("“”)

pg:keepalive()
end

function callback(topic, payload)
ngx.log(ngx.ERR, "a data reveived topic:", topic)
--write to db
local data = "">
ngx.log(ngx.ERR, "a data reveived data:", data["devEUI"])

local ok, err = ngx.timer.at(0, wt_nodedata, data)
ngx.log(ngx.ERR, "1 create timer: ", ok) -----------------------------------1 监听到消息时这里还可以看到打印日志
ngx.log(ngx.ERR, "2 create timer: ", err)
if not ok then
ngx.log(ngx.ERR, "failed to create timer: ", err)
return
end
end

local MQTT = require "mqtt"
MQTT.Utility.set_debug(true)
MQTT.client.KEEP_ALIVE_TIME=60
local mqtt_client = MQTT.client.create('127.0.0.1', 1883, callback)
local error_message = mqtt_client:connect(tostring(ngx.worker.pid()))

local topics = {"application/+/rx"}
mqtt_client:subscribe(topics)

while (error_message == nil) do
error_message = mqtt_client:handler() -----------------------------------0 监听到消息时这里会调用callback函数
socket.sleep(1.0)
end

--

YuanSheng Wang

---------------------------------------

My Github: https://github.com/membphis

OpenResty lover ^_^

xieqin1

我用的这个：

https://github.com/iotrac/mqtt-lua

应该也是阻塞的？

改成cosocket版本的话.. 有什么建议么

我看这个连postgresql的：https://github.com/leafo/pgmoon

他这里面就支持luasocket和cosocket两种，可否按他这样改？或者我在mqtt-lua中直接require pgmoon的socket模块

在 2016年8月23日星期二 UTC+8上午11:49:01，YuanSheng写道：

刚刚看了一下你用的 lua 库，是不是这个？ https://github.com/geekscape/mqtt_lua/blob/master/lua/mqtt_library.lua

这个库是阻塞运行的。

openresty 自身的异步世界，是基于 cosocket 的，与 luasocket 区别很大。

luasocket 是阻塞执行，他会破坏 nginx 的异步世界，让所有的行为 stop。

这也就合理解释了 ngx.timer.at 没有执行的原因，因为 openresty 被 mqtt 阻塞了。

建议方法：

对 https://github.com/geekscape/mqtt_lua/blob/master/lua/mqtt_library.lua 动手术，修改成 cosocket 版本。

On Mon, Aug 22, 2016 at 7:06 PM, <xie...@gmail.com> wrote:
Hi，我在init_worker中监听mqtt，收到数据则通过ngx.timer.at写数据库（postgresql），目前监听成功，但是ngx.timer.at总调用不到，不知各位有何建议？代码如下：

local function wt_nodedata(premature, data)
ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"]) ---------------------------2 监听到消息时这里无打印日志了
local pg = pgmoon.new({
host = "127.0.0.1",
port = "5432",
database = "postgre1",
user = "admin",
password = "dbpassword"
})

pg:connect()

local res = pg:query("“”)

pg:keepalive()
end

function callback(topic, payload)
ngx.log(ngx.ERR, "a data reveived topic:", topic)
--write to db
local data = "">
ngx.log(ngx.ERR, "a data reveived data:", data["devEUI"])

local ok, err = ngx.timer.at(0, wt_nodedata, data)
ngx.log(ngx.ERR, "1 create timer: ", ok) -----------------------------------1 监听到消息时这里还可以看到打印日志
ngx.log(ngx.ERR, "2 create timer: ", err)
if not ok then
ngx.log(ngx.ERR, "failed to create timer: ", err)
return
end
end

local MQTT = require "mqtt"
MQTT.Utility.set_debug(true)
MQTT.client.KEEP_ALIVE_TIME=60
local mqtt_client = MQTT.client.create('127.0.0.1', 1883, callback)
local error_message = mqtt_client:connect(tostring(ngx.worker.pid()))

local topics = {"application/+/rx"}
mqtt_client:subscribe(topics)

while (error_message == nil) do
error_message = mqtt_client:handler() -----------------------------------0 监听到消息时这里会调用callback函数
socket.sleep(1.0)
end

--

--

YuanSheng Wang
---------------------------------------
My Github: https://github.com/membphis
OpenResty lover ^_^

xieqin1

试了下貌似cosocket在init_worker阶段是无法创建的，，，？调用ngx.socket.tcp()提示：

API disabled in the context of init_worker_by_lua*

在 2016年8月23日星期二 UTC+8上午11:49:01，YuanSheng写道：

刚刚看了一下你用的 lua 库，是不是这个？ https://github.com/geekscape/mqtt_lua/blob/master/lua/mqtt_library.lua

这个库是阻塞运行的。

openresty 自身的异步世界，是基于 cosocket 的，与 luasocket 区别很大。

luasocket 是阻塞执行，他会破坏 nginx 的异步世界，让所有的行为 stop。

这也就合理解释了 ngx.timer.at 没有执行的原因，因为 openresty 被 mqtt 阻塞了。

建议方法：

对 https://github.com/geekscape/mqtt_lua/blob/master/lua/mqtt_library.lua 动手术，修改成 cosocket 版本。

On Mon, Aug 22, 2016 at 7:06 PM, <xie...@gmail.com> wrote:
Hi，我在init_worker中监听mqtt，收到数据则通过ngx.timer.at写数据库（postgresql），目前监听成功，但是ngx.timer.at总调用不到，不知各位有何建议？代码如下：

local function wt_nodedata(premature, data)
ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"]) ---------------------------2 监听到消息时这里无打印日志了
local pg = pgmoon.new({
host = "127.0.0.1",
port = "5432",
database = "postgre1",
user = "admin",
password = "dbpassword"
})

pg:connect()

local res = pg:query("“”)

pg:keepalive()
end

function callback(topic, payload)
ngx.log(ngx.ERR, "a data reveived topic:", topic)
--write to db
local data = "">
ngx.log(ngx.ERR, "a data reveived data:", data["devEUI"])

local ok, err = ngx.timer.at(0, wt_nodedata, data)
ngx.log(ngx.ERR, "1 create timer: ", ok) -----------------------------------1 监听到消息时这里还可以看到打印日志
ngx.log(ngx.ERR, "2 create timer: ", err)
if not ok then
ngx.log(ngx.ERR, "failed to create timer: ", err)
return
end
end

local MQTT = require "mqtt"
MQTT.Utility.set_debug(true)
MQTT.client.KEEP_ALIVE_TIME=60
local mqtt_client = MQTT.client.create('127.0.0.1', 1883, callback)
local error_message = mqtt_client:connect(tostring(ngx.worker.pid()))

local topics = {"application/+/rx"}
mqtt_client:subscribe(topics)

while (error_message == nil) do
error_message = mqtt_client:handler() -----------------------------------0 监听到消息时这里会调用callback函数
socket.sleep(1.0)
end

--

--

YuanSheng Wang
---------------------------------------
My Github: https://github.com/membphis
OpenResty lover ^_^

xieqin1

难道要我先用一个timer来启动监听mqtt 然后再用另一个timer来写postgresql? 我屮艸芔茻太不优雅了吧。。。

在 2016年8月23日星期二 UTC+8上午11:49:01，YuanSheng写道：

刚刚看了一下你用的 lua 库，是不是这个？ https://github.com/geekscape/mqtt_lua/blob/master/lua/mqtt_library.lua

这个库是阻塞运行的。

openresty 自身的异步世界，是基于 cosocket 的，与 luasocket 区别很大。

luasocket 是阻塞执行，他会破坏 nginx 的异步世界，让所有的行为 stop。

这也就合理解释了 ngx.timer.at 没有执行的原因，因为 openresty 被 mqtt 阻塞了。

建议方法：

对 https://github.com/geekscape/mqtt_lua/blob/master/lua/mqtt_library.lua 动手术，修改成 cosocket 版本。

On Mon, Aug 22, 2016 at 7:06 PM, <xie...@gmail.com> wrote:
Hi，我在init_worker中监听mqtt，收到数据则通过ngx.timer.at写数据库（postgresql），目前监听成功，但是ngx.timer.at总调用不到，不知各位有何建议？代码如下：

local function wt_nodedata(premature, data)
ngx.log(ngx.ERR, " aaaaaaaaaaa1111111 timer: ", data["mac"]) ---------------------------2 监听到消息时这里无打印日志了
local pg = pgmoon.new({
host = "127.0.0.1",
port = "5432",
database = "postgre1",
user = "admin",
password = "dbpassword"
})

pg:connect()

local res = pg:query("“”)

pg:keepalive()
end

function callback(topic, payload)
ngx.log(ngx.ERR, "a data reveived topic:", topic)
--write to db
local data = "">
ngx.log(ngx.ERR, "a data reveived data:", data["devEUI"])

local ok, err = ngx.timer.at(0, wt_nodedata, data)
ngx.log(ngx.ERR, "1 create timer: ", ok) -----------------------------------1 监听到消息时这里还可以看到打印日志
ngx.log(ngx.ERR, "2 create timer: ", err)
if not ok then
ngx.log(ngx.ERR, "failed to create timer: ", err)
return
end
end

local MQTT = require "mqtt"
MQTT.Utility.set_debug(true)
MQTT.client.KEEP_ALIVE_TIME=60
local mqtt_client = MQTT.client.create('127.0.0.1', 1883, callback)
local error_message = mqtt_client:connect(tostring(ngx.worker.pid()))

local topics = {"application/+/rx"}
mqtt_client:subscribe(topics)

while (error_message == nil) do
error_message = mqtt_client:handler() -----------------------------------0 监听到消息时这里会调用callback函数
socket.sleep(1.0)
end

--

--

YuanSheng Wang
---------------------------------------
My Github: https://github.com/membphis
OpenResty lover ^_^