用 Openresty + Websocket + Redis Stream 实现一个简单聊天室，保持长连接，并通过Redis Stream 转发消息

在 Openresty + Websocket + Redis 实现一个简单聊天室，保持长连接，并通过Redis转发消息 中，详细实现了一个简单的聊天室，将消息发送到 Redis 队列中，并使用 Redis 订阅功能，将消息推送给所有连接的 WebSocket 客户端。

但是Redis订阅消息时，read_reply() 会阻塞 Redis 连接，使其无法被其他请求复用，这在实际使用中会带来问题。

改进方案有：

  • 使用 Redis Pub/Sub + 共享连接池模式
  • 使用 Redis Lists + 轮询模式
  • 使用 Redis Streams，stream 获取消息时可以是非阻塞的

下面是一个 Redis Streams 方案的例子：

redis stream lua 实现

将消息发送到 Redis Streams 中，然后订阅 Redis Streams，然后从 Redis Streams 中获取消息响应客户端。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250

-- 简易聊天室
local server = require "resty.websocket.server"
local redis = require "resty.redis"
local cjson = require "cjson"

local uri = ngx.var.uri
ngx.log(ngx.INFO, "uri: ", uri)
local channel_name = "user:"
local uname = uri

-- 创建 websocket 连接
local wb, err = server:new{
  timeout = 30000,  -- 增加到30秒
  max_payload_len = 65535
}

-- 输出连接ID到error.log
ngx.log(ngx.INFO, "WS connected, ID: ", ngx.var.connection)

if not wb then
  ngx.log(ngx.ERR, "failed to create new websocket: ", err)
  return ngx.exit(444)
end

local function get_redis()
    local red = redis:new()
    red:set_timeout(5000)  -- 5秒超时
    
    -- 仅当没有活跃连接时创建新连接
    local ok, err = red:connect("172.17.0.3", 6379)
    -- ngx.log(ngx.INFO, "---------- connect ", ok, err)
    if not ok then
        ngx.log(ngx.ERR, "failed to connect redis: ", err)
        return nil, err
    end
    -- ok, err = red:auth("123456")
    -- ok, err = red:select(1)
    return red
end

-- 消息发布方式，使用 Redis Streams
local push = function()
    local last_id = "$"  -- 从最新消息开始
    local poll_interval = 0.5
    
    -- 为每个连接创建唯一的消费者名称
    -- local consumer_name = "consumer_" .. ngx.var.connection .. "_" .. ngx.time()

    while not wb.fatal do
        ngx.sleep(poll_interval)
        
        local red, err = get_redis()
        if red then
            -- red:set_timeout(2000)
            
            -- 使用 XREAD 读取新消息（修正参数格式）
            local res, err = red:xread("COUNT", 1, "BLOCK", 100, "STREAMS", "chat_stream", last_id)
            ngx.log(ngx.INFO, "----- Received messages from stream res: ", cjson.encode(res), "err:", err)
            if res and type(res) == "table" then
                ngx.log(ngx.INFO, "Received messages from stream res: ", cjson.encode(res))
                for _, stream_data in ipairs(res) do
                    local stream_name = stream_data[1]
                    local messages = stream_data[2]
                    ngx.log(ngx.INFO, "Received messages from stream: ", stream_name)
                    ngx.log(ngx.INFO, "Received messages: ", cjson.encode(messages))
                    
                    if type(messages) == "table" then
                        for _, message_data in ipairs(messages) do
                            ngx.log(ngx.INFO, "Received message_data: ", cjson.encode(message_data))
                            local message_id = message_data[1]
                            local fields = message_data[2]
                            ngx.log(ngx.INFO, "Received message_id: ", message_id)
                            ngx.log(ngx.INFO, "Received fields: ", cjson.encode(fields))
                            
                            -- 从字段中提取消息内容
                            local message = nil
                            if type(fields) == "table" then
                                -- 遍历字段键值对找到 message 内容
                                for i = 1, #fields, 2 do
                                    if fields[i] == "message" and i + 1 <= #fields then
                                        message = fields[i + 1]
                                        break
                                    end
                                end
                            end
                            
                            ngx.log(ngx.INFO, "Extracted message: ", message)
                            
                            -- 发送消息到客户端
                            if not wb.fatal and message then
                                local bytes, send_err = wb:send_text(message)
                                if not bytes then
                                    ngx.log(ngx.ERR, "Failed to send message: ", send_err)
                                    break
                                end
                                ngx.log(ngx.INFO, "Successfully sent message to client: ", message)
                            end
                            
                            -- 更新最后读取的ID
                            last_id = message_id
                            ngx.log(ngx.INFO, "Last message ID: ", last_id)
                        end
                    end
                end
            elseif err and not string.find(err:lower(), "timeout") then
                ngx.log(ngx.ERR, "Redis stream read error: ", err)
            end
            
            -- 返回连接池
            local ok, keep_err = red:set_keepalive(10000, 100)
            ngx.log(ngx.INFO, "red to set keepalive: ", ok, keep_err)
            if not ok then
                ngx.log(ngx.WARN, "Failed to set keepalive: ", keep_err)
                -- 清理资源
                pcall(function() red:close() end)
            end
        end
    end
end

-- 启用一个线程用来发送信息
local co = ngx.thread.spawn(push)

-- 主线程
local last_ping_time = ngx.now()
local ping_interval = 25  -- 25秒发送一次ping

while true do
    -- 如果连接损坏 退出
    if wb.fatal then
        ngx.log(ngx.ERR, "WebSocket connection is fatal: ", wb.fatal)
        break
    end

    local data, typ, err = wb:recv_frame()
    ngx.log(ngx.INFO, "---------- recv_frame ", err, " data: ", data, " typ: ", typ)

    if not data then
        -- 检查是否是超时错误（包括各种形式的超时）
        if err then
            local is_timeout = (err == "timeout" or string.find(err:lower(), "timeout"))
            
            if is_timeout then
                -- 超时情况下发送心跳
                ngx.log(ngx.DEBUG, "WebSocket recv timeout, checking ping interval")
                if ngx.now() - last_ping_time > ping_interval then
                    local bytes, ping_err = wb:send_ping()
                    ngx.log(ngx.INFO, "send ping: ", bytes, " err: ", ping_err)
                    if not bytes then
                      ngx.log(ngx.WARN, "failed to send ping: ", ping_err)
                      -- 如果发送ping失败，可能连接已断开
                      if ping_err and not string.find(ping_err:lower(), "timeout") then
                          break
                      end
                    end
                    last_ping_time = ngx.now()
                end
                -- 超时是正常的，继续循环
            else
                -- 非超时错误，记录并退出
                ngx.log(ngx.ERR, "failed to receive frame: ", err)
                break
            end
        else
            -- err 为 nil，这可能表示连接正常关闭
            ngx.log(ngx.INFO, "WebSocket connection closed normally")
            break
        end
    elseif typ == "close" then
        ngx.log(ngx.INFO, "ws closed by client, connection ID: ", ngx.var.connection)
        break
    elseif typ == "ping" then
        -- 回复心跳
        local bytes, pong_err = wb:send_pong()
        -- ngx.log(ngx.INFO, "----------- send_pong", pong_err)
        if not bytes then
            ngx.log(ngx.ERR, "failed to send pong: ", pong_err)
            break
        end
        ngx.log(ngx.INFO, "client ping received, WS connected, ID: ", ngx.var.connection)
    elseif typ == "pong" then
        -- 心跳回包
        ngx.log(ngx.INFO, "client ponged")
        last_ping_time = ngx.now()  -- 重置ping时间
    elseif typ == "text" then
        ngx.log(ngx.INFO, "received text message: ", uname .. data)
        -- 将消息发送到 redis 频道
        local red2, err = get_redis()
        -- ngx.log(ngx.INFO, "----------- get_redis2 ----", err)
        if not red2 then
            ngx.log(ngx.ERR, "failed to connect redis: ", err)
            break
        end

        -- 发布到 Redis Stream 而不是 Pub/Sub
        -- 使用 XADD 发布消息到 Stream
        local res, pub_err = red2:xadd("chat_stream", "*", 
            "sender", uname, "message", data, "timestamp", tostring(ngx.time()))
        -- ngx.log(ngx.INFO, "----------- xadd result ----", res)
        -- ngx.log(ngx.INFO, "----------- xadd error ----", pub_err)
        if not res then
            ngx.log(ngx.ERR, "failed to add to redis stream: ", pub_err)
        else
            ngx.log(ngx.INFO, "Successfully added message to stream, ID: ", res)
        end

        -- 关键！将连接放回池中
        local ok, keepalive_err = red2:set_keepalive(10000, 100)
        -- ngx.log(ngx.INFO, "----------- set_keepalive ----", keepalive_err)
        if not ok then
            ngx.log(ngx.WARN, "failed to set keepalive: ", keepalive_err)
            -- 即使放回连接池失败，也要关闭连接
            pcall(function() red2:close() end)
        end
    else
        -- 处理其他类型的消息
        ngx.log(ngx.INFO, "received frame type: ", typ, " data: ", data)
    end
end

wb:send_close()

-- 等待推送线程结束，但设置超时避免永久等待
local ok, err = ngx.thread.kill(co, 5)  -- 5秒超时
if not ok then
    ngx.log(ngx.ERR, "failed to kill thread: ", err)
end

-- 清理资源
local red, err = get_redis()
if red then
    -- 删除 Stream 
    local res, err = red:del({"chat_stream"})
    if not res then
        ngx.log(ngx.ERR, "failed to delete stream: ", err)
    else
        ngx.log(ngx.INFO, "deleted stream: ", res)
    end

    local ok, keepalive_err = red:set_keepalive(10000, 100)
    -- ngx.log(ngx.INFO, "----------- set_keepalive ----", keepalive_err)
    if not ok then
        ngx.log(ngx.WARN, "failed to set keepalive: ", keepalive_err)
        -- 即使放回连接池失败，也要关闭连接
        pcall(function() red:close() end)
    end
end

ngx.log(ngx.INFO, "WS closed, ID: ", ngx.var.connection)

客户端html

简单的websocket客户端demo

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

<!DOCTYPE HTML>
<html>

<head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0, user-scalable=no">
    <style>
        p{margin:0;}
    </style>
    <script src="https://code.jquery.com/jquery-3.3.1.min.js"></script>
    <script type="text/javascript">
        var ws = null;
        var pingInterval = null; // 用于保存定时器ID

        function WebSocketConn() {
            if (ws != null && ws.readyState == 1) {
                log("已经在线");
                return
            }

            if ("WebSocket" in window) {
                // Let us open a web socket
                ws = new WebSocket("ws://127.0.0.1/ws/stream/user001");

                ws.onopen = function () {
                    log('成功进入聊天室');
                    // 启动定时发送ping消息
                    startPing();
                };

                ws.onmessage = function (event) {
                    log(event.data)
                };

                ws.onclose = function () {
                    // websocket is closed.
                    log("已经和服务器断开");
                    stopPing(); // 停止ping
                };

                ws.onerror = function (event) {
                    console.log("error " + event.data);
                    stopPing(); // 出错时也停止ping
                };
            } else {
                // The browser doesn't support WebSocket
                alert("WebSocket NOT supported by your Browser!");
            }
        }

        function SendMsg() {
            if (ws != null && ws.readyState == 1) {
                var msg = document.getElementById('msgtext').value;
                ws.send(msg);
            } else {
                log('请先进入聊天室');
            }
        }

        function WebSocketClose() {
            if (ws != null && ws.readyState == 1) {
                ws.close();
                log("发送断开服务器请求");
            } else {
                log("当前没有连接服务器")
            }
        }

        // 新增：启动定时发送ping
        function startPing() {
            if (pingInterval) clearInterval(pingInterval);
            pingInterval = setInterval(function() {
                if (ws != null && ws.readyState == 1) {
                    ws.send('ping');
                }
            }, 10000); // 每30秒发送一次ping
        }

        // 新增：停止定时发送ping
        function stopPing() {
            if (pingInterval) {
                clearInterval(pingInterval);
                pingInterval = null;
            }
        }

        function log(text) {
            var li = document.createElement('p');
            li.appendChild(document.createTextNode(text));
            //document.getElementById('log').appendChild(li);
            $("#log").prepend(li);
            return false;
        }

        WebSocketConn();
    </script>
</head>

<body>
<div id="sse">
    <a href="javascript:WebSocketConn()">进入聊天室</a> &nbsp;
    <a href="javascript:WebSocketClose()">离开聊天室</a>
    <br>
    <br>
    <input id="msgtext" type="text">
    <br>
    <a href="javascript:SendMsg()">发送信息</a>
    <br>
    <br>
    <div id="log"></div>
</div>
</body>

</html>

浏览器中打开测试页面，进入聊天室，输入信息，点击发送信息，即可看到信息发送到聊天室。

openresty 配置

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

    location /ws/stream {
        access_log logs/ws_access.log ws_log;  # 专用WS日志文件
        # WebSocket 支持
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_read_timeout 300s;
        
        content_by_lua_file lua/websocket_stream.lua;
    }

golang redis stream 测试示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/redis/go-redis/v9"
)

// Writing to a Stream in Go
func main() {
	rdb := redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "", // 没有密码，默认值
		DB:       0,  // 默认DB 0
	})

	var ctx = context.Background()

	// Reading from a Stream in Go
	go func() {
		for {
			streams, err := rdb.XRead(ctx, &redis.XReadArgs{
				Streams: []string{"events", "$"},
				Block:   0, // 阻塞毫秒数，没有设置就是非阻塞模式
				Count:   1, // 数量
			}).Result()
			if err != nil {
				log.Fatalf("XRead failed: %v", err)
			}

			for _, stream := range streams {
				for _, msg := range stream.Messages {
					fmt.Printf("ID: %s, Values: %v\n", msg.ID, msg.Values)
				}
			}
		}
	}()

	streamKey := "events"

	ticker := time.NewTicker(10 * time.Second)
	for range ticker.C {
		// 使用 XADD 向队列添加消息
		// XADD key ID field value [field value ...]
		// XADD events * name Alice action login time xxx
		args := &redis.XAddArgs{
			Stream: streamKey,
			MaxLen: 1000,
			Approx: true,
			Values: map[string]interface{}{
				"user":   "Alice",
				"action": "login",
				"time":   time.Now().Format(time.RFC3339),
			},
		}
		id, err := rdb.XAdd(ctx, args).Result()
		if err != nil {
			log.Fatalf("XAdd failed: %v", err)
		}

		fmt.Printf("Written to stream with ID: %s\n", id)
	}
}

参考