openresty lua用Redis的Stream解决消息订阅问题

使用 Redis Streams 解决消息订阅和消费的问题，可以避免在订阅模式下的连接管理问题。下面是如何使用 OpenResty 和 Redis Streams 实现类似的功能。

配置 nginx.conf

确保你的 nginx.conf 文件中配置了 Lua 模块和 Redis 集群的连接信息：

http {lua_shared_dict redis_cluster_slot_locks 10m;lua_shared_dict redis_cluster_slot_cache 10m;init_worker_by_lua_file /path/to/init_worker.lua;server {listen 8080;location /publish {content_by_lua_block {local redis_cluster = require "resty.rediscluster"local config = {name = "testCluster",serv_list = {{ ip = "127.0.0.1", port = 7000 },{ ip = "127.0.0.1", port = 7001 },{ ip = "127.0.0.1", port = 7002 },{ ip = "127.0.0.1", port = 7003 },{ ip = "127.0.0.1", port = 7004 },{ ip = "127.0.0.1", port = 7005 }},keepalive_timeout = 60000,keepalive_cons = 1000,connection_timout = 1000,max_redirection = 5}local red = redis_cluster:new(config)local res, err = red:xadd("mystream", "*", "message", "Hello, World!")if not res thenngx.say("failed to publish: ", err)returnendngx.say("message published to stream mystream")local ok, err = red:set_keepalive(10000, 100)if not ok thenngx.say("failed to set keepalive: ", err)returnend}}}
}

init_worker.lua

在 init_worker.lua 中编写消费逻辑，并确保在消费模式下正确管理连接：

local redis_cluster = require "resty.rediscluster"
local config = {name = "testCluster",serv_list = {{ ip = "127.0.0.1", port = 7000 },{ ip = "127.0.0.1", port = 7001 },{ ip = "127.0.0.1", port = 7002 },{ ip = "127.0.0.1", port = 7003 },{ ip = "127.0.0.1", port = 7004 },{ ip = "127.0.0.1", port = 7005 }},keepalive_timeout = 60000,keepalive_cons = 1000,connection_timout = 1000,max_redirection = 5
}local function consume_stream(premature)if premature thenreturnendlocal red = redis_cluster:new(config)local last_id = "0"  -- 开始读取的起始 IDwhile true dolocal res, err = red:xread("COUNT", 10, "BLOCK", 1000, "STREAMS", "mystream", last_id)if not res thenngx.log(ngx.ERR, "failed to read stream: ", err)breakendif res and res[2] thenfor _, stream in ipairs(res[2]) dofor _, message in ipairs(stream[2]) dolocal id = message[1]local fields = message[2]ngx.log(ngx.INFO, "received message: ", fields[2])last_id = idendendendendred:close()-- Schedule another stream consumption attemptlocal ok, err = ngx.timer.at(1, consume_stream)if not ok thenngx.log(ngx.ERR, "failed to create timer: ", err)end
end-- Schedule the consume_stream function
local ok, err = ngx.timer.at(0, consume_stream)
if not ok thenngx.log(ngx.ERR, "failed to create timer: ", err)
end

关键点解释

1. 避免在消费模式下调用 set_keepalive：

   • 在消费模式下，我们不会尝试将连接放入连接池，而是直接读取消息并处理。

2. 连接管理：

   • 消费操作在 consume_stream 函数内进行。
   • 在读取失败时，记录错误并退出循环，然后释放连接。

3. 使用 ngx.timer.at 调度消费函数：

   • 使用 ngx.timer.at(0, consume_stream) 调度消费函数，以便在 worker 初始化时立即开始消费。

4. 处理 Redis Streams：

   • 使用 xread 命令读取流中的消息。
   • 通过循环读取消息并处理。

测试

1. 启动 OpenResty 并配置上述 nginx.conf 和 init_worker.lua。

2. 使用发布接口发布消息，查看 OpenResty 日志确认消息已接收：

   curl "http://localhost:8080/publish"
   

   在 OpenResty 日志中应该看到类似如下输出：

   2024/07/09 16:04:00 [info] 12345#0: *1 [lua] init_worker.lua:23: received message: Hello, World!
   

通过这些配置和代码，你可以在 OpenResty 中使用 Redis Streams 实现对消息的长期消费，并正确处理连接的生命周期和错误。