redis发布和订阅(pub/sub) - foundwant - 博客园 主要是 ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context

发布和订阅模式是常用和很方便的模式，下面记录redis中对pub/sub的支持；

复制代码; "复制代码")

Pub/Sub: "发布/订阅"在redis中，被设计的非常轻量级和简洁，它做到了消息的“发布”和“订阅”的

基本能力;但是尚未提供关于消息的持久化等各种企业级的特性。

一个Redis client发布消息,其他多个redis client订阅消息,发布的消息“即发即失”,redis
不会持久保存发布的消息;消息订阅者也将只能得到订阅之后的消息,通道中此前的消息将无
从获得。

消息发布者，即publish客户端，无需独占链接，你可以在publish消息的同时，使用同一个redis-client链接进行其他操作（例如：INCR等）
消息订阅者，即subscribe客户端，需要独占链接，即进行subscribe期间，redis-client无法穿插其他操作，
此时client以阻塞的方式等待“publish端”的消息；因此这里subscribe端需要使用单独的链接，甚至需要在额外的线程中使用。  
Tcp默认连接时间固定，如果在这时间内sub端没有接收到pub端消息，或pub端没有消息产生，sub端的连接都会被强制回收，  
这里就需要使用特殊手段解决，用定时器来模拟pub和sub之间的保活机制，定时器时间不能超过TCP最大连接时间，具体根据机器环境来定；  
一旦subscribe端断开链接，将会失去部分消息，即链接失效期间的消息将会丢失，所以这里就需要考虑到借助redis的list来持久化；  
如果你非常关注每个消息，那么你应该基于Redis做一些额外的补充工作，如果你期望订阅是持久的，那么如下的设计思路可以借鉴： 1) subscribe端:
    首先向一个Set集合中增加“订阅者ID”, 此Set集合保存了“活跃订阅”者,
    订阅者ID标记每个唯一的订阅者,此Set为 "活跃订阅者集合"  
2) subcribe端开启订阅操作,并基于Redis创建一个以 "订阅者ID" 为KEY的LIST数据结构,
    此LIST中存储了所有的尚未消费的消息,此List称为 "订阅者消息队列"
    
3) publish端:
    每发布一条消息之后,publish端都需要遍历 "活跃订阅者集合",并依次
    向每个 "订阅者消息队列" 尾部追加此次发布的消息. 4) 到此为止,我们可以基本保证,发布的每一条消息,都会持久保存在每个 "订阅者消息队列" 中. 5) subscribe端,每收到一个订阅消息,在消费之后,必须删除自己的 "订阅者消息队列" 头部的一条记录. 6) subscribe端启动时,如果发现自己的 "订阅者消息队列" 有残存记录, 那么将会首先消费这些记录,然后再去订阅.

以上方法可以保证成功到达的消息必消费不丢失；
但还是会存在ngx业务机方自丢失数据问题，也就是ngx业务机自身问题或网络问题导致ngx业务机发布的消息没有送达redis机器；
更完善的确认机制才能彻底解决上述存在问题；

注意，在实际ngx_lua_redis应用中，redis单个客户端订阅模式下仅能使用有限的几个命令，不能使用其它结构命令，如lpop,rpush等；
因为 publish是普通的request/response模式, 但subscribe不是，否则会报错：
　　　　　　ERR only (P)SUBSCRIBE \/ (P)UNSUBSCRIBE \/ PING \/ QUIT allowed in this cont
关于这点以下是官网一般解释:

 You are required to use two connections for pub and sub. A subscriber connection cannot issue any commands

other than subscribe, psubscribe, unsubscribe, punsubscribe (although @Antirez has hinted of a subscriber-safe
ping in the future). If you try to do anything else, redis tells you: -ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context
(note that you can't test this with redis-cli, since that understands the protocol well enough to prevent you
from issuing commands once you have subscribed - but any other basic socket tool should work fine)

This is because subscriber connections work very differently \- rather than working on a request/response basis,

incoming messages can now come in at any time, unsolicited.
publish is a regular request/response command, so must be sent on a regular connection, not a subscriber connection.

于是该特性不适用单例模式，要解决上面局限，需要多客户端辅助操作同一结果，下列代码中会有展示；

复制代码; "复制代码")

下面示例是ngx_lua_redis生产环境下实验结果，有兴趣的可以分析

复制代码; "复制代码")

--[[ cosocket即coroutine+socket

顺序执行,但它是非阻塞执行方式
因为nginx core是非阻塞执行;
redis中subscribe是阻塞方式,
因此在nginx_lua平台中使用redis
中sub特性无法保持阻塞连接状态;
流程模型:http://www.cnblogs.com/foundwant/p/6382083.html \]\]

local args = ngx.req.get_uri_args() local ttype = args.type -- pub/sub

local function newRedis(timeout, ip, port, section) local red = redis.new()

red:set_timeout(timeout) local ok, err = red:connect(ip, port) if not ok then nlog.dinfo("connect:" .. err) end red:select(section) return red end

local red = newRedis(10000, "127.0.0.1", "6379", 0) local bak = newRedis(10000, "127.0.0.1", "6379", 0) local function subscribe(channel) local res, err = red:subscribe(channel) if not res then nlog.dinfo("subscribe error.") return nil, err end

--这里以函数返回,不然sub会在这里断连失去可操作性
--这就是提到的特殊之一
local function read\_func(do\_read) if nil == do_read or true == do_read then res, err = red:read_reply() if not res then
            return nil, err end

        return res end red:unsubscribe(channel)
    red:set_keepalive(60000, 100) --连接回收

bak:close()

    bak:set_keepalive(60000, 100) --断连后重启等待
    red = newRedis(10000, "127.0.0.1", "6379", 0)
    red:subscribe(channel)

    bak = newRedis(10000, "127.0.0.1", "6379", 0) return
end

return read_func end

local subset = "subset" --set
local channel = "test" --list
consume = function(length) --若订阅者消息队列有残余,先消费,再订阅

    for i=1, llength do
        local recv, err = red:lpop(channel)  --头部开始消费
        nlog.dinfo("recv:" .. cjson.encode(recv)) end redis\_util.coroutine\_count = 1
    coroutine.yield() end

--订阅者
if "sub" == ttype then

--向set集合增加"订阅者id"

red:sadd(subset, channel) --为每个"订阅者id"建立list

local llength = red:llen(channel) if 0 == llength then red:rpush(channel, "hello") else
    --若订阅者消息队列有残余,先消费,再订阅
    for i=1, llength do
        local recv, err = red:lpop(channel)  --头部开始消费
        nlog.dinfo("recv:" .. cjson.encode(recv)) end
end nlog.dinfo("run coroutine after...") --开始订阅
local func, err = subscribe(channel) while true do
    local res, err = func()   --res:\["message","test","world"\] 
    if err then func(false) end
    --在redis的订阅模式中,
    --单例模式下只能使用固定几个命令\[ (P)SUBSCRIBE,(P)UNSUBSCRIBE,QUIT,PING,... \],
    --无法使用其它命令,比如lpop, rpush等命令,
    --所以这里无法使用red:lpop()来执行出队删除操作,
    --只能另起一个客户端对象来进行删除操作;
    local oo, ooerr = bak:lpop(channel) 
    nlog.dinfo("bak lpop:" .. cjson.encode(oo))
    nlog.dinfo("res:" .. cjson.encode(res))
    ngx.sleep(1) end

end

--发布者,测试用,实际调用是在业务层
if "pub" == ttype then

--先发布,再追加队列
--local subchannel, err = red:spop(subset)
--nlog.dinfo("subchannel:" .. type(subchannel))
--if "userdata" ~= type(subchannel) then
for i=1, 1000 do
    local str = "world_" .. i
    red:publish(channel, str)
    red:rpush(channel, str) --尾部追加
    ngx.sleep(0.1) end
--end

end

--监听器,crontab定时运行
if "spy" == ttype then

while true do red:publish(channel, "0")
    ngx.sleep(60) end

end ok, err = red:set_keepalive(60000, 100) if not ok then ngx.say("set_keepalive:", err) end ngx.print("rpush done.")
ngx.exit(200)

复制代码; "复制代码")

Lua_Redis

原网址: 访问
创建于: 2019-04-08 01:09:30
目录: default
标签: 无

未标明原创文章均为采集，版权归作者所有，转载无需和我联系，请注明原出处，南摩阿彌陀佛，知识，不只知道，要得到

请先后发表评论
  • 最新评论
  • 总共0条评论