极简群聊：OpenResty + Redis，把语音也塞进 WebSocket

flowchart LR
  A(浏览器) -->|文字：/ws/chat · JSON 帧| G[OpenResty 网关]
  A -->|语音：/ws/voice · 二进制帧| G
  G -->|PUBLISH 到群频道| R[(Redis)]
  R -->|每个 worker 一条订阅| G
  G -->|按群转发给对应连接| A

              Redis Pub/Sub
                    |
          +---------+---------+
          |         |         |
        worker1   worker2   worker3
          |         |         |
        队列      队列      队列
          |         |         |
        WS: A, B    WS: C, D    WS: E, F

sequenceDiagram
    participant A as Alice（浏览器）
    participant G as OpenResty
    participant R as Redis Pub/Sub
    participant B as Bob（浏览器）
    A->>A: AudioWorklet 攒 20ms PCM（640 字节）
    A->>G: /ws/voice 二进制帧（贴 sender、group）
    G->>R: PUBLISH group:1001:voice
    R-->>G: 各 worker 订阅器收到
    G->>B: 从 B 的 /ws/voice 推过去（发送者的那一路不回）
    B->>B: 抖动缓冲 + AudioContext 播放

lua_package_path "/usr/local/openresty/lualib/?.lua;;";

init_worker_by_lua_block {
    local subscriber = require "subscriber"
    subscriber.start()          -- 每个 worker 一条 Redis 订阅
}

server {
    listen 443 ssl;
    ssl_certificate     /etc/nginx/certs/fullchain.pem;
    ssl_certificate_key /etc/nginx/certs/privkey.pem;

    location /ws/chat {
        content_by_lua_file /app/lua/chat.lua;     -- 文字
    }
    location /ws/voice {
        content_by_lua_file /app/lua/voice.lua;    -- 语音
    }
    location / {
        root /app/web;
        index index.html;
    }
}

local redis   = require "resty.redis"
local clients = require "clients"

local function loop()
    local red = redis:new()
    red:set_timeout(0)                       -- 订阅要长超时
    local ok, err = red:connect("redis", 6379)
    if not ok then return nil, err end
    red:psubscribe("group:*")                -- 订阅所有群频道
    while true do
        local reply = red:read_reply()       -- 阻塞等消息
        if type(reply) == "table" and reply[1] == "pmessage" then
            clients.route(reply[3], reply[4])   -- 按频道转发
        end
    end
end

function _M.start()
    ngx.timer.at(0, function()
        local ok, err = pcall(loop)
        if not ok then ngx.log(ngx.ERR, "subscriber: ", err) end
    end)
end

return _M

local server  = require "resty.websocket.server"
local clients = require "clients"
local message = require "message"
local cjson   = require "cjson.safe"

local wb, err = server:new({ timeout = 1000, max_payload_len = 8192 })
if not wb then return ngx.exit(444) end

local me = clients.register("chat", ngx.var.auth_user_id, ngx.var.request_id)

local function flush()
    while true do
        local item = me:pop()
        if not item then return true end
        local ok = wb:send_text(item.data)   -- 文字连接只推 JSON 文本
        if not ok then return nil, "push failed" end
    end
end

while true do
    local data, typ, err = wb:recv_frame()
    if typ == "close" then break end
    if typ == "text" then
        local msg = cjson.decode(data)
        if msg then
            if msg.type == "send_message" then
                message.send(me.user_id, msg.payload.group_id, msg.payload.text)
            elseif msg.type == "join_group" then
                clients.join(me, msg.payload.group_id)
            elseif msg.type == "heartbeat" then
                clients.refresh(me)          -- 续命 presence:connX
            end
        end
    elseif err and err ~= "timeout" then
        break
    end
    if not flush() then break end
end

clients.unregister(me)
wb:send_close()

local redis = require "resty.redis"

local _M = {}

function _M.get()
    local red = redis:new()
    red:set_timeout(1000)
    local ok, err = red:connect("redis", 6379)
    if not ok then ngx.log(ngx.ERR, "redis: ", err) return nil, err end
    return red
end

function _M.release(red)
    if red then red:set_keepalive(10000, 100) end
end

return _M

local redis  = require "redis"
local cjson  = require "cjson.safe"

local function too_long(text)
    return type(text) ~= "string" or #text == 0 or #text > 4096
end

function _M.send(user_id, group_id, text)
    if too_long(text) then return nil, "invalid message" end

    local red, err = redis.get()
    if not red then return nil, err end

    if red:sismember("group:" .. group_id .. ":members", user_id) ~= 1 then
        redis.release(red)
        return nil, "not a group member"
    end

    -- 先落 Stream（留历史、供补发），拿到消息号再广播
    local id, err = red:xadd("group:" .. group_id .. ":msg",
                             "MAXLEN", "~", 1000, "*", "payload",
                             cjson.encode({ group_id = group_id,
                                            sender_id = user_id,
                                            text = text, t = ngx.time() }))
    if not id then redis.release(red) return nil, err end

    red:publish("group:" .. group_id .. ":msg",
                cjson.encode({ type = "message", message_id = id,
                               group_id = group_id, sender_id = user_id,
                               text = text }))
    redis.release(red)
    return id
end

return _M

-- 语音群在握手时由 JWT 决定并校验，连接一开就知道待在哪个群里
local server        = require "resty.websocket.server"
local clients       = require "clients"
local redis         = require "redis"

local wb, err = server:new({ timeout = 1000, max_payload_len = 65536 })
if not wb then return ngx.exit(444) end

local group_id = ngx.var.auth_voice_group
local me = clients.register("voice", ngx.var.auth_user_id, ngx.var.request_id, group_id)

local function flush()
    while true do
        local item = me:pop()
        if not item then return true end
        local ok = wb:send_binary(item.data)   -- 语音连接只推二进制帧
        if not ok then return nil, "push failed" end
    end
end

while true do
    local data, typ, err = wb:recv_frame()
    if typ == "close" then break end
    if typ == "binary" then
        -- 贴 sender 的小二进制头：1 字节名字长度 + 名字 + 帧数据
        local frame = string.char(#me.user_id) .. me.user_id .. data

        local red, e = redis.get()
        if red then
            red:publish("group:" .. group_id .. ":voice", frame)  -- 广播，Redis 不挑内容
            redis.release(red)
        end
    elseif typ == "text" then
        -- 应用级 ping/pong：探测半开连接，比只听 TCP 关闭靠谱
        if data == '{"type":"ping"}' then wb:send_text('{"type":"pong"}') end
    elseif err and err ~= "timeout" then
        break
    end
    if not flush() then break end
end

clients.unregister(me)
wb:send_close()

// 两条连接，各发各的：文字走 /ws/chat，语音走 /ws/voice
// （token 由服务端签发，语音那支的 token 里带着群号）
const wsChat  = new WebSocket("/ws/chat?token="  + token);
const wsVoice = new WebSocket("/ws/voice?token=" + token);

const sendText = (group_id, text) =>
  wsChat.send(JSON.stringify({ type: "send_message",
                               payload: { group_id, text } }));

// 采集：AudioWorklet 每 20ms 给 320 个 float32（16kHz）
class Mic extends AudioWorkletProcessor {
  process(inputs) {
    const ch = inputs[0][0];
    if (ch) this.port.postMessage(ch.slice());
    return true;
  }
}
registerProcessor("mic", Mic);

// 发送：float32 转 int16 后，走 /ws/voice 的二进制帧
micNode.port.onmessage = (e) => wsVoice.send(float32ToInt16(e.data).buffer);

// 接收：文字上屏走 wsChat；语音帧攒两帧再播，就是 40ms 的抖动缓冲
wsChat.onmessage = (e) => appendMessage(JSON.parse(e.data));

const q = [];
wsVoice.binaryType = "arraybuffer";
wsVoice.onmessage = (e) => {
  if (e.data instanceof ArrayBuffer) {
    q.push(e.data);
    if (q.length >= 2) {                 // 至少攒两帧，稳一点
      schedulePlayback(q.shift());
    }
  }
};

@startmindmap
<style>
mindmapDiagram {
  node {
    BackgroundColor #F8F9FA
    RoundCorner 10
    Padding 10
    FontSize 13
  }
  :depth(0) {
    BackgroundColor #1E3A5F
    FontColor white
    FontSize 18
    FontStyle bold
  }
  :depth(1) {
    FontSize 15
    FontStyle bold
  }
  :depth(2) {
    FontSize 13
  }
}
</style>

* 极简群聊
** 为什么媒体塞不进 WebSocket
*** 浏览器只认 ICE/DTLS/SRTP
*** OpenResty 终结不了加密会话
** 两条路
*** 真 WebRTC + SFU：三进程、UDP、强
*** 语音当消息：两进程、443、简
** 一张网关两条连接
*** /ws/chat 文字 JSON 帧
*** /ws/voice 语音二进制帧
*** 每 worker 一条 Redis 订阅
*** 有界队列挡慢客户端
** Redis 三样东西
*** 成员 Set
*** 在线 TTL
*** 文字 Stream
*** 语音只广播不落盘
** 一帧语音的一生
*** 20ms / 640B / 256kbps
*** Opus 省一个数量级
*** 时延约 60-110ms
** 放弃与容量
*** 回声、弱网表现
*** 权衡：带宽 vs 帧率
** 升级梯
*** Opus → 静音检测 → SFU
*** 升级是加进程不是推翻
@endmindmap