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

  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

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

local channel_name = "chat"
local uname = "网友" .. tostring(math.random(10,99)) .. ": "

-- 创建 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
    return red
end

local push = function()
    -- 创建redis连接
    local red, err = get_redis()
    ngx.log(ngx.INFO, "---------- get redis ", err)
    if not red then
        ngx.log(ngx.ERR, "failed to connect redis: ", err)
        wb:send_close()
        return
    end

    -- -- 设置适当的超时时间
    -- red:set_timeout(30000)  -- 30秒超时
    -- 设置较短的超时时间，有利连接复用
    red:set_timeout(1000)  -- 1秒超时

    --订阅聊天频道
    local res, err = red:subscribe(channel_name)
    ngx.log(ngx.INFO, "---------- subscribe ", err)
    if not res then
        ngx.log(ngx.ERR, "failed to sub redis: ", err)
        wb:send_close()
        red:set_keepalive(10000, 100)
        return
    end

    -- 死循环获取消息
    while true do
        -- 检查 WebSocket 连接是否还存活
        if wb.fatal then
            ngx.log(ngx.INFO, "WebSocket connection closed, exiting push thread")
            break
        end
        
        local res, err = red:read_reply()
        ngx.log(ngx.INFO, "---------- read_reply ", err)
        if res then
            local item = res[3]
            ngx.log(ngx.INFO, "read text: ", uname .. item)
            local bytes, err = wb:send_text(item)
            ngx.log(ngx.INFO, "---------- send_text ", err)
            if not bytes then
                -- 错误直接退出
                ngx.log(ngx.ERR, "failed to send text: ", err)
                break
            else
                ngx.log(ngx.INFO, "sent text: ", uname .. bytes)
            end
        else
            -- 特别处理超时情况
            if err == "timeout" then
                -- 超时是正常的，继续循环
                ngx.log(ngx.DEBUG, "Redis read timeout, continuing...")
            else
                -- 其他错误需要退出
                ngx.log(ngx.ERR, "failed to read text: ", err)
                break
            end
        end
    end
    
    wb:send_close()
    -- 清理资源
    pcall(function()
        red:unsubscribe(channel_name)
        red:set_keepalive(10000, 100)
    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
        local res, pub_err = red2:publish(channel_name, uname .. data)
        ngx.log(ngx.INFO, "----------- publish ----", pub_err)
        if not res then
            ngx.log(ngx.ERR, "failed to publish redis: ", pub_err)
        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
ngx.log(ngx.INFO, "WS closed, ID: ", ngx.var.connection)

  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");

                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>