请教：在lua-resty-redis中如何使用pub/sub实现多key订阅和多值返回呢？

duwenbin71 · 2014-01-14T12:57:31+00:00

Hello! 2014/1/13 nikua...: > 如果在所有worker内共享该TCP连接的话，&#x...

请教：在lua-resty-redis中如何使用pub/sub实现多key订阅和多值返回呢？

duwenbin71

目前我只能使用red:subscribe多次订阅，使用red:read_reply阻塞等待返回，但返回值只有一个key的消息，其实我同时向所有订阅的key中都发布了消息。

agentzh

Hello!

2014/1/13  duwenbin71:
> 目前我只能使用red:subscribe多次订阅，使用red:read_reply阻塞等待返回，但返回值只有一个key的消息，其实我同时向所有订阅的key中都发布了消息。
>

lua-resty-redis 库只是忠实地返回了 Redis 服务器的响应。

根据 Redis pubsub 的官方手册，推送消息貌似都是一条一条地进行的：

    http://redis.io/topics/pubsub

所以你的 subscriber 需要继续调用 read_reply() 接收其他 key 对应的消息。

Regards,
-agentzh

duwenbin71

thanks agentzh.

我目前的实现策略正如您所说，通过多次调用read_reply()获取所有消息。但这样做必须将timeout时间缩短到10秒内，不然通知延迟周期太长。

如果能实现一个类似于WinAPI中的WaitForMultipleObjects这类的api貌似更好？

不管怎样，谢谢了。

在 2014年1月14日星期二UTC+8下午1时17分34秒，agentzh写道：

Hello!

2014/1/13 duwenbin71:
> 目前我只能使用red:subscribe多次订阅，使用red:read_reply阻塞等待返回，但返回值只有一个key的消息，其实我同时向所有订阅的key中都发布了消息。
>

lua-resty-redis 库只是忠实地返回了 Redis 服务器的响应。

根据 Redis pubsub 的官方手册，推送消息貌似都是一条一条地进行的：

http://redis.io/topics/pubsub

所以你的 subscriber 需要继续调用 read_reply() 接收其他 key 对应的消息。

Regards,
-agentzh

agentzh

Hello!

2014/1/13 wb du:
> 我目前的实现策略正如您所说，通过多次调用read_reply()获取所有
> 消息。但这样做必须将timeout时间缩短到10秒内，不然通知延迟周
> 期太长。

我不明白为什么你这里需要 timeout. 当你使用 subscribe 命令订阅多个频道时，redis
难道不会把各个频道上发布的新消息及时推送给你吗？你为什么需要自己阻塞等待呢？

> 如果能实现一个类似于WinAPI中的WaitForMultipleObjects这类的api貌似更好？
>

Redis 的 pub/sub API
就应该是这种模式。难道你的用法有错误？你能提供一个最小化的完整的可以复现问题的代码示例吗？（完整的标准是在其他人的机器上可以很容易地运行你的示例，比如我的机器。）

Best regards,
-agentzh

duwenbin71

基本的思路是：

1. 前端发起多频道订阅请求

2. lua循环完成订阅

3. 使用read_reply阻塞1秒

4. 如得到通知，保存到table中，回到3继续阻塞

5. 超时返回时，有几种情况：

1）在上次循环中已获得通知，跳到6；

2）未获得通知，如还在1分钟内则回到3继续阻塞；

3）未获得通知，超出1分钟跳到6；

6. 取消所有订阅

7. table转为json返回给前端。

下面是我的代码：

WebAPI:

GET http://localhost/api/sub?chns=msg,news

前端JS：

function push() {
	$.ajax({
		url : '/api/sub?chns=msg,news',
		success : function(data) {
				show(data);
				push();
		},
		error : function() {
			push();
		},
		dataType : 'json'
	});
}

Lua：

local red = redis:new()
red:set_timeout(1000)
local ok, err = red:connect('127.0.0.1',6379)

local chns = ngx.var.arg_chns
local channels = {}
for i, channel in pairs(split(chns, ',')) do
	channels[i] = channel
	local result, err = red:subscribe(channel)
end

local ret = {}
local second=0 
local found = false
--60秒内循环
while second<60 do
	result, err = red:read_reply() --阻塞1秒
	second = second + 1
	--当上次循环中得到消息，本次循环无消息时向客户端返回
	if found and result==nil then
		break
	end
	--保存消息到ret中
	if result~=nil and result[1]=="message" then
		ret[result[2]] = result[3]
		found = true
	end
end
--取消订阅
for i, channel in pairs(channels) do
	red:unsubscribe(chn)
end 
ngx.print(json.encode(ret))
red:close()

在 2014年1月15日星期三UTC+8上午5时52分18秒，agentzh写道：

Hello!

2014/1/13 wb du:
> 我目前的实现策略正如您所说，通过多次调用read_reply()获取所有
> 消息。但这样做必须将timeout时间缩短到10秒内，不然通知延迟周
> 期太长。

我不明白为什么你这里需要 timeout. 当你使用 subscribe 命令订阅多个频道时，redis
难道不会把各个频道上发布的新消息及时推送给你吗？你为什么需要自己阻塞等待呢？

> 如果能实现一个类似于WinAPI中的WaitForMultipleObjects这类的api貌似更好？
>

Redis 的 pub/sub API
就应该是这种模式。难道你的用法有错误？你能提供一个最小化的完整的可以复现问题的代码示例吗？（完整的标准是在其他人的机器上可以很容易地运行你的示例，比如我的机器。）

Best regards,
-agentzh

doujiang24

我的理解：

subscribe 客户端跟 redis 服务器建立了长链接，订阅的频道上有消息，都会立即发送到客户端，不管监听了多少个频道，每次一条

read_reply 是客户端从链接上读取消息，如果有消息会立马返回，timeout 是当前读操作的最大等待时间（不是 tcp 长链接的超时）

长链接一直存在，所以只要有订阅频道有消息， read_reply 总是能及时返回，timeout 时间长一点也不会造成消息延时

在 2014年1月15日上午9:25，wb du <duwe...@gmail.com>写道：

基本的思路是：
1. 前端发起多频道订阅请求
2. lua循环完成订阅
3. 使用read_reply阻塞1秒

4. 如得到通知，保存到table中，回到3继续阻塞
5. 超时返回时，有几种情况：
1）在上次循环中已获得通知，跳到6；
2）未获得通知，如还在1分钟内则回到3继续阻塞；
3）未获得通知，超出1分钟跳到6；
6. 取消所有订阅
7. table转为json返回给前端。

下面是我的代码：
WebAPI:
GET http://localhost/api/sub?chns=msg,news

前端JS：
function push() { $.ajax({ url : '/api/sub?chns=msg,news', success : function(data) { show(data); push(); }, error : function() { push(); }, dataType : 'json' }); }

Lua：
local red = redis:new() red:set_timeout(1000) local ok, err = red:connect('127.0.0.1',6379) local chns = ngx.var.arg_chns local channels = {} for i, channel in pairs(split(chns, ',')) do channels[i] = channel local result, err = red:subscribe(channel) end local ret = {} local second=0 local found = false --60秒内循环 while second<60 do result, err = red:read_reply() --阻塞1秒 second = second + 1 --当上次循环中得到消息，本次循环无消息时向客户端返回 if found and result==nil then break end --保存消息到ret中 if result~=nil and result[1]=="message" then ret[result[2]] = result[3] found = true end end --取消订阅 for i, channel in pairs(channels) do red:unsubscribe(chn) end ngx.print(json.encode(ret)) red:close()

在 2014年1月15日星期三UTC+8上午5时52分18秒，agentzh写道：
Hello!

2014/1/13 wb du:
> 我目前的实现策略正如您所说，通过多次调用read_reply()获取所有
> 消息。但这样做必须将timeout时间缩短到10秒内，不然通知延迟周
> 期太长。

我不明白为什么你这里需要 timeout. 当你使用 subscribe 命令订阅多个频道时，redis
难道不会把各个频道上发布的新消息及时推送给你吗？你为什么需要自己阻塞等待呢？

> 如果能实现一个类似于WinAPI中的WaitForMultipleObjects这类的api貌似更好？
>

Redis 的 pub/sub API
就应该是这种模式。难道你的用法有错误？你能提供一个最小化的完整的可以复现问题的代码示例吗？（完整的标准是在其他人的机器上可以很容易地运行你的示例，比如我的机器。）

Best regards,
-agentzh

--

agentzh

Hello!

2014/1/14 wb du:
> 基本的思路是：
>     1. 前端发起多频道订阅请求
>     2. lua循环完成订阅
[...]
> local channels = {}
> for i, channel in pairs(split(chns, ',')) do
> channels[i] = channel
> local result, err = red:subscribe(channel)
> end

你这里订阅多个频道的逻辑是错误的，难怪你会引入完全不必要的延时和超时错误。

根据 Redis 的官方手册，同时订阅多个频道是通过单个 subscribe 命令来实现的。比如同时订阅 foo 和 bar
这两个频道，应使用单个 subscribe 命令

    red:subscribe('foo', 'bar')

如果你频道在一个 Lua table 里，则可以这么写：

    red:subscribe(unpack(channels))

在你随后读取的各条消息中会由服务器标识当前消息属于哪一个频道。

更多细节请参见 Redis 官网上的 pubsub 文档：

    http://redis.io/topics/pubsub

Best regards,
-agentzh

duwenbin71

red:subscribe(unpack(channels)) 替换了循环订阅的方式，实际上两种写法的作用是一样的。

我现在面临的问题是：同时订了两个频道，并且两个频道的消息几乎是同时到达时，read_reply只能返回一个，不得已的情况下，只好采用循环read_reply的方式来获取第二个消息。

在 2014年1月15日星期三UTC+8下午12时12分38秒，agentzh写道：

Hello!

2014/1/14 wb du:
> 基本的思路是：
> 1. 前端发起多频道订阅请求
> 2. lua循环完成订阅
[...]
> local channels = {}
> for i, channel in pairs(split(chns, ',')) do
> channels[i] = channel
> local result, err = red:subscribe(channel)
> end

你这里订阅多个频道的逻辑是错误的，难怪你会引入完全不必要的延时和超时错误。

根据 Redis 的官方手册，同时订阅多个频道是通过单个 subscribe 命令来实现的。比如同时订阅 foo 和 bar
这两个频道，应使用单个 subscribe 命令

red:subscribe('foo', 'bar')

如果你频道在一个 Lua table 里，则可以这么写：

red:subscribe(unpack(channels))

在你随后读取的各条消息中会由服务器标识当前消息属于哪一个频道。

更多细节请参见 Redis 官网上的 pubsub 文档：

http://redis.io/topics/pubsub

Best regards,
-agentzh

duwenbin71

是。

多频道订阅，且多消息同时到达的情况需要重新考虑。

在 2014年1月15日星期三UTC+8上午10时19分02秒，doujiang写道：

我的理解：

subscribe 客户端跟 redis 服务器建立了长链接，订阅的频道上有消息，都会立即发送到客户端，不管监听了多少个频道，每次一条

read_reply 是客户端从链接上读取消息，如果有消息会立马返回，timeout 是当前读操作的最大等待时间（不是 tcp 长链接的超时）

长链接一直存在，所以只要有订阅频道有消息， read_reply 总是能及时返回，timeout 时间长一点也不会造成消息延时

在 2014年1月15日上午9:25，wb du <duwen...@gmail.com>写道：

基本的思路是：
1. 前端发起多频道订阅请求
2. lua循环完成订阅
3. 使用read_reply阻塞1秒

4. 如得到通知，保存到table中，回到3继续阻塞
5. 超时返回时，有几种情况：
1）在上次循环中已获得通知，跳到6；
2）未获得通知，如还在1分钟内则回到3继续阻塞；
3）未获得通知，超出1分钟跳到6；
6. 取消所有订阅
7. table转为json返回给前端。

下面是我的代码：
WebAPI:
GET http://localhost/api/sub?chns=msg,news

前端JS：
function push() { $.ajax({ url : '/api/sub?chns=msg,news', success : function(data) { show(data); push(); }, error : function() { push(); }, dataType : 'json' }); }

Lua：
local red = redis:new() red:set_timeout(1000) local ok, err = red:connect('127.0.0.1',6379) local chns = ngx.var.arg_chns local channels = {} for i, channel in pairs(split(chns, ',')) do channels[i] = channel local result, err = red:subscribe(channel) end local ret = {} local second=0 local found = false --60秒内循环 while second<60 do result, err = red:read_reply() --阻塞1秒 second = second + 1 --当上次循环中得到消息，本次循环无消息时向客户端返回 if found and result==nil then break end --保存消息到ret中 if result~=nil and result[1]=="message" then ret[result[2]] = result[3] found = true end end --取消订阅 for i, channel in pairs(channels) do red:unsubscribe(chn) end ngx.print(json.encode(ret)) red:close()

在 2014年1月15日星期三UTC+8上午5时52分18秒，agentzh写道：
Hello!

2014/1/13 wb du:
> 我目前的实现策略正如您所说，通过多次调用read_reply()获取所有
> 消息。但这样做必须将timeout时间缩短到10秒内，不然通知延迟周
> 期太长。

我不明白为什么你这里需要 timeout. 当你使用 subscribe 命令订阅多个频道时，redis
难道不会把各个频道上发布的新消息及时推送给你吗？你为什么需要自己阻塞等待呢？

> 如果能实现一个类似于WinAPI中的WaitForMultipleObjects这类的api貌似更好？
>

Redis 的 pub/sub API
就应该是这种模式。难道你的用法有错误？你能提供一个最小化的完整的可以复现问题的代码示例吗？（完整的标准是在其他人的机器上可以很容易地运行你的示例，比如我的机器。）

Best regards,
-agentzh

--