关于websocket server接收客户端消息的写法的问题

kytexzy

首先声明，我对websocket认知为0, 刚刚拜读了https://medium.com/p/1778601c9e05
，有两点不明：
1.是不是所有的websocket服务都需要一个

while true do
...

end
式的无限循环，这样会不会影响服务效率.

2.如果在这个循环中用redis的blpop订阅消息，这个blpop会不会阻塞当前的循环

--
with kind regards

kytexzy

我的用例这样写：
....
local blocked = false
....
local function blpop()
    local res, err = red:blpop('wsmsg', 100)

    if res then
      local bytes, err = wb:send_text(table.concat(res, ':'))

      if not bytes then
        ngx.log(ngx.ERR, "failed to send text: ", err)
        return ngx.exit(444)
      end
    elseif not res then
        ngx.log(ngx.ERR, "failed to blpop wsmsg", err)
    end

    return false
end

while true do
local data, typ, err = wb:recv_frame()
if wb.fatal then
    ngx.log(ngx.ERR, "failed to receive frame: ", err)
    return ngx.exit(444)
end
if not data then
    local bytes, err = wb:send_ping()
    if not bytes then
      ngx.log(ngx.ERR, "failed to send ping: ", err)
      return ngx.exit(444)
    end
elseif typ == "close" then
    break
elseif typ == "text" then
    local bytes, err = wb:send_text(data)
    if not bytes then
      ngx.log(ngx.ERR, "failed to send text: ", err)
      return ngx.exit(444)
    end
end
ngx.log(ngx.CRIT, 'over a loop', ':', typ, ':', data)

if not blocked then
    blocked = true
    blocked = blpop()
end
end
wb:send_close()
这样在websocket服务中用blpop订阅消息不知道合适不合适, 请指教。

On Tuesday, January 7, 2014 4:36:20 PM UTC+8, Nero.Ping wrote:

首先声明，我对websocket认知为0, 刚刚拜读了https://medium.com/p/1778601c9e05
，有两点不明：
1.是不是所有的websocket服务都需要一个
while true do
...
end
式的无限循环，这样会不会影响服务效率.

2.如果在这个循环中用redis的blpop订阅消息，这个blpop会不会阻塞当前的循环
--
with kind regards

agentzh

Hello!

2014/1/7 Zheng Ping:
> ，有两点不明：
> 1.是不是所有的websocket服务都需要一个
> while true do
> ...
> end
> 式的无限循环，

不一定。这取决于你的业务流程。那篇博客里提供的只是一个 demo 示例.

> 这样会不会影响服务效率.
>

这里的 while true 并不会阻塞 nginx worker 进程，因为里面的非阻塞的 IO 操作会让当前的“轻线程”经常被切出去。

> 2.如果在这个循环中用redis的blpop订阅消息， 这个blpop会不会阻塞当前的循环

会。因为 blpop 是同步操作，用在这个 websocket
的分发循环里会让这个循环长时间挂起。当然，这也取决于你的业务流程。如果业务流程并不要求在 blpop 等待时在当前的 websocket
连接上同时处理其他数据包，则让你的 websocket 读取和处理循环挂起是很自然很正确的做法。否则你应当使用自己的“轻线程”把 blpop
操作做成相对于你的主循环的异步操作：

    https://github.com/chaoslawful/lua-nginx-module#ngxthreadspawn

建议总是交待你的业务流程和需求，而不是简单地询问具体做法是否合理。毕竟需求不同，适合的具体做法也会大不相同。

Regards,
-agentzh

richard.chen

在“轻线程”中，他这个 blpop() 函数中调用wb:send_text 是不是需要全双工的websocket才行啊，现在是不能工作的吧？

2014/1/8 Yichun Zhang (agentzh) <age...@gmail.com>

Hello!

2014/1/7 Zheng Ping:

> ，有两点不明：
> 1.是不是所有的websocket服务都需要一个
> while true do
> ...
> end
> 式的无限循环，

不一定。这取决于你的业务流程。那篇博客里提供的只是一个 demo 示例.

> 这样会不会影响服务效率.
>

这里的 while true 并不会阻塞 nginx worker 进程，因为里面的非阻塞的 IO 操作会让当前的“轻线程”经常被切出去。

> 2.如果在这个循环中用redis的blpop订阅消息，这个blpop会不会阻塞当前的循环

会。因为 blpop 是同步操作，用在这个 websocket
的分发循环里会让这个循环长时间挂起。当然，这也取决于你的业务流程。如果业务流程并不要求在 blpop 等待时在当前的 websocket
连接上同时处理其他数据包，则让你的 websocket 读取和处理循环挂起是很自然很正确的做法。否则你应当使用自己的“轻线程”把 blpop
操作做成相对于你的主循环的异步操作：

https://github.com/chaoslawful/lua-nginx-module#ngxthreadspawn

建议总是交待你的业务流程和需求，而不是简单地询问具体做法是否合理。毕竟需求不同，适合的具体做法也会大不相同。

Regards,
-agentzh

--

agentzh

Hello!

2014/1/9 Chen Richard:
> 在“轻线程”中，他这个 blpop() 函数中调用wb:send_text 是不是需要全双工的websocket才行啊，现在是不能工作的吧？
>

当多个轻级程可能会同时操作一个 cosocket 对象时，目前需要通过 lua-resty-lock 这样的锁进行并发保护：

    https://github.com/agentzh/lua-resty-lock

近期会通过一个补丁允许一个“轻线程”写 cosocket 的同时，另一个“轻线程”读取同一个 cosocket.
但如果两个“轻线程”会同时写或者同时读一个 cosocket 时还是需要进行锁保护。

当然，lua-resty-lock 是可以跨 worker 进程共享的锁，而这里也可以简单地使用 Lua 变量而不是共享内存字典来作锁。

Regards,
-agentzh

kytexzy

我的项目是一个多人在线协作完成任务的文档生成器。当多人同时编辑一个文档时, 要一个人编辑的文档能实时地让另外同时在线的人能马上看得到, 我觉得用redis的pub/sub可能比blpop更合适。
我看了看看
https://groups.google.com/forum/#!msg/openresty/K7GJ9i-3svI/GpbBRSkcgf0J
上面关于game server的描述，我感觉是应该用ngx.thread.spawn来订阅消息。
我有一个疑问，是不是当我向某个频道发布消息时也应该对该频道(channel)上锁(resty-lock)来阻止其它线程可能的同时写操作, 那么多用户订阅消息的时候有不有必要对该频道加锁?如果订阅消息的时候也要加锁，如果我此刻就想向该频道pub消息需要怎么处理呢?

On Wednesday, January 8, 2014 5:35:14 AM UTC+8, agentzh wrote:

Hello!

2014/1/7 Zheng Ping:
> ，有两点不明：
> 1.是不是所有的websocket服务都需要一个
> while true do
> ...
> end
> 式的无限循环，

不一定。这取决于你的业务流程。那篇博客里提供的只是一个 demo 示例.

> 这样会不会影响服务效率.
>

这里的 while true 并不会阻塞 nginx worker 进程，因为里面的非阻塞的 IO 操作会让当前的“轻线程”经常被切出去。

> 2.如果在这个循环中用redis的blpop订阅消息，这个blpop会不会阻塞当前的循环

会。因为 blpop 是同步操作，用在这个 websocket
的分发循环里会让这个循环长时间挂起。当然，这也取决于你的业务流程。如果业务流程并不要求在 blpop 等待时在当前的 websocket
连接上同时处理其他数据包，则让你的 websocket 读取和处理循环挂起是很自然很正确的做法。否则你应当使用自己的“轻线程”把 blpop
操作做成相对于你的主循环的异步操作：

https://github.com/chaoslawful/lua-nginx-module#ngxthreadspawn

建议总是交待你的业务流程和需求，而不是简单地询问具体做法是否合理。毕竟需求不同，适合的具体做法也会大不相同。

Regards,
-agentzh

agentzh

Hello!

2014/1/18 Nero.Ping:
> 我有一个疑问，是不是当我向某个频道发布消息时也应该对该频道(channel)上锁(resty-lock)来阻止其它线程可能的同时写操作,
> 那么多用户订阅消息的时候有不有必要对该频道加锁?如果订阅消息的时候也要加锁，如果我此刻就想向该频道pub消息需要怎么处理呢?
>

是这样的，我们需要首先区分客户端和服务器端。

对于服务器端，Redis 自身会有串行化的保护。对于客户端，只要不会出现多个“轻线程”同时读写同一个 cosocket 对象（即同一个到
Redis 的 TCP 连接），就不需要锁。频道是一个抽象的概念，本身是由 Redis 服务器端来保证并发下的一致性的。

Regards,
-agentzh

kytexzy

我现在有个问题，我参考 https://medium.com/p/1778601c9e05 写了一个最简单的用resty.redis订阅消息, 并用websocket发送到客户段的例子, 读取订阅的消息也是用的while true循环, 订阅消息的函数用ngx.thread.spawn了一个轻线程, 可是问题出来了，在openresty 1.4.3.9中我的代码木有问题，但是放到1.5.8.1中不能正常运行, 我不知道问题出在哪里，请哪位高人帮我看看,我不知道是我的写法不对还是其它原因。

我把我的最小化测试代码上传附件。

另外我在我的项目里面用了resty.lock也是在1.4.3.9木有问题, 但是放到1.5.8.1就跑不起来, 当运行在openresty 1.5.8.1中时，一旦建立websocket链接后, 我按F5刷浏览器，浏览器总是迟迟不能刷新.

Attachment: t4.lua
Description: Binary data

agentzh

Hello!

2014-02-04 Nero.Ping:
> 订阅消息的函数用ngx.thread.spawn了一个轻线程, 可是问题出来了，在openresty
> 1.4.3.9中我的代码木有问题，但是放到1.5.8.1中不能正常运行,

“不能正常运行”具体是指什么现象？你在 nginx 的错误日志文件里是否看到什么错误消息？

>
> 我把我的最小化测试代码上传附件。
>

你只提供了服务器端的代码，请同时提供你在客户端使用的代码，以及你复现问题所需的具体步骤，这样我可以尝试在我本地运行你的例子 :)

> 另外我在我的项目里面用了resty.lock也是在1.4.3.9木有问题, 但是放到1.5.8.1就跑不起来, 当运行在openresty
> 1.5.8.1中时，一旦建立websocket链接后, 我按F5刷浏览器，浏览器总是迟迟不能刷新.
>

对于 lua-resty-lock 的问题，也请提供一个最小化的例子 :)

多谢！

Best regards,
-agentzh

kytexzy

On Wednesday, February 5, 2014 2:38:14 AM UTC+8, Nero.Ping wrote:

我现在有个问题，我参考 https://medium.com/p/1778601c9e05 写了一个最简单的用resty.redis订阅消息, 并用websocket发送到客户段的例子, 读取订阅的消息也是用的while true循环, 订阅消息的函数用ngx.thread.spawn了一个轻线程, 可是问题出来了，在openresty 1.4.3.9中我的代码木有问题，但是放到1.5.8.1中不能正常运行, 我不知道问题出在哪里，请哪位高人帮我看看,我不知道是我的写法不对还是其它原因。

我把我的最小化测试代码上传附件。

我现在把客户端的代码也上传上来，见附件.

我的nginx配置文件是:

# vim:set ft= sw=4 ts=4 et:

# for valgrind debug
#daemon off;
#master_process off;

worker_processes 2;

# dump core dump, for production, it is not needed
#worker_rlimit_core 50M;


events {
    worker_connections 1024;
}

http {
    #error_log /home/zhen/temp/nginx/openresty_error.log debug;
    error_log logs/ngx_error.log debug;

    types {
        text/html                             html htm shtml;
        text/css                              css;
        application/x-_javascript_              js;
    }

    default_type application/octet-stream;
    sendfile on;

    lua_package_path '$prefix/lua/?.lua;;';
    lua_package_cpath '$prefix/lua/lib/?.so;;';
    lua_max_pending_timers 4096;
    lua_max_running_timers 1024;
    lua_socket_log_errors off;
    lua_check_client_abort on;
    lua_shared_dict ws_locks 100k;

    lua_code_cache off;

    init_by_lua '
        require "resty.core";
    ';

    server {
        listen 8090 so_keepalive=2s:2s:8;

        root ../;

        location / {
            deny all;
        }

        location  /t {
            content_by_lua_file lua/t.lua;
        }

        location /t1 {
            content_by_lua_file lua/t/t1.lua;
        }

        location /t4 {
            content_by_lua_file lua/t/t4.lua;
        }
    }
}

我的服务端环境是 nginx 1.4.2 做前端代理服务器, 后端的业务服务器是openresty 1.5.8.1，
在浏览器中打开这个测试页面127.0.0.1/test-ws.html, test-ws.html直接用websocket连接到27.0.0.1:

8090/t4，当我第一次连接到websocket后用redis-cli直接publish my-channel msg, 此时redis客户端显示：
redis 127.0.0.1:6379> publish my_channel msg
(integer) 0
然后刷新浏览器,浏览器就迟迟不能刷新, 我的log文件太大有225M, 我也不好上传.

但是如果我用openresty 1.4.3.9则可以成功把我上传的例子跑起来, 我不太懂到底是哪里有问题。

Attachment: t4.lua
Description: Binary data

agentzh

Hello!

2014-02-05 Nero.Ping:
> 然后刷新浏览器,浏览器就迟迟不能刷新, 我的log文件太大有225M, 我也不好上传.
>

你的 log 文件里面主要是什么级别的错误消息？[error]？[debug]?检几条典型的样本贴出来看看？

我约摸知道你这里是什么问题了。你的 t4.lua 中有一个没有终止条件的无限循环：

    while true do
        res, err = red:read_reply()

        if res then
            local msg = cjson.encode(res)

            table.insert(buf, msg)
        end
    end

这个循环没有终止条件。你应该对 read_reply() 调用进行正确的错误处理，即 res 为 nil 的情形。

对于 openresty 1.4.3.9 中的 0.17 版的 lua-resty-redis 库而言，它允许在 read_reply()
返回 timeout 超时错误时继续再次调用 read_reply() 来重试。而对于 openresty 1.5.8.1 中的
lua-resty-redis 0.18，read_reply() 会在超时错误发生时立即关闭当前的 redis 连接，于是后续紧跟着的
read_reply() 调用都会返回 nil, "closed" 这两个返回值。于是你上面那个循环就热循环了，而且是死循环。

我刚刚在最新的 lua-resty-redis 0.19 给 read_reply() 方法重新添加了在读超时发生后允许直接重试
read_reply() 的支持。该版本的 lua-resty-redis 已经包含在了 ngx_openresty 1.5.9.1rc2
预发布版中：

    http://openresty.org/download/ngx_openresty-1.5.9.1rc2.tar.gz

另外，你的 t4.lua 中的那个有问题的死循环应该修改成下面这个样子：

    while true do
        res, err = red:read_reply()
        if not res and err ~= "timeout" then
            ngx.log(ngx.ERR, "redis connection error: ", err)
            -- well, you can re-connect to redis via red:connect() and
            -- re-subscribe here instead of return directly
            return
        end

        if res then
            local msg = cjson.encode(res)
            table.insert(buf, msg)
        end
    end

一般地，对于 while true 这样的循环应当特别小心，千万不要在某一个代码路径上热循环了。

Regards,
-agentzh

kytexzy

对于 openresty 1.4.3.9 中的 0.17 版的 lua-resty-redis 库而言，它允许在 read_reply()
返回 timeout 超时错误时继续再次调用 read_reply() 来重试。而对于 openresty 1.5.8.1 中的
lua-resty-redis 0.18，read_reply() 会在超时错误发生时立即关闭当前的 redis 连接，于是后续紧跟着的
read_reply() 调用都会返回 nil, "closed" 这两个返回值。于是你上面那个循环就热循环了，而且是死循环。

我刚刚在最新的 lua-resty-redis 0.19 给 read_reply() 方法重新添加了在读超时发生后允许直接重试
read_reply() 的支持。该版本的 lua-resty-redis 已经包含在了 ngx_openresty 1.5.9.1rc2
预发布版中：

http://openresty.org/download/ngx_openresty-1.5.9.1rc2.tar.gz

另外，你的 t4.lua 中的那个有问题的死循环应该修改成下面这个样子：

while true do
res, err = red:read_reply()
if not res and err ~= "timeout" then
ngx.log(ngx.ERR, "redis connection error: ", err)
-- well, you can re-connect to redis via red:connect() and
-- re-subscribe here instead of return directly
return
end

if res then
local msg = cjson.encode(res)
table.insert(buf, msg)
end
end

一般地，对于 while true 这样的循环应当特别小心，千万不要在某一个代码路径上热循环了。

春哥有理，我使用openresty 1.5.9后就ok了, 这个超时(timeout)后重试的特性太好了, 不过代码中加上

        if not res and err ~= "timeout" then

            ngx.log(ngx.ERR, "redis connection error: ", err)

            -- well, you can re-connect to redis via red:connect() and

            -- re-subscribe here instead of return directly

            return

        end

err ~= "timeout"
的确更严谨