基于Erlang Cowboy与WebSocket实现群聊：连接存储与消息转发方案

核心需求 #

实现基于Cowboy的WebSocket聊天室，任意客户端发送的消息可同步转发至所有其他已接入的连接，达到群组聊天效果。

现有实现的缺口 #

当前代码仅实现了单连接的消息计数回复，缺少两个核心能力：

• 已接入WebSocket连接的统一存储/索引能力
• 单消息触发全连接广播的转发逻辑

现有业务代码 #

-module(chat_conn).

-export([ init/2
    , websocket_init/1
    , websocket_handle/2
    , websocket_info/2
    , terminate/3
    ]).

-include("chat.hrl").

init(Req, [WsConnIdleTimeout]) ->
    ?LOG_DEBUG("[CHAT-CONN] New HTTP Request on: /api , Pid: ~p", [self()]),
    WsOpts = #{idle_timeout => WsConnIdleTimeout},
    State = #{counter => 0},
    {cowboy_websocket, Req, State, WsOpts}.

websocket_init(State0) ->
    ?LOG_DEBUG("[FOOZI-CONN] HTTP Upgraded to WebSocketm pid: ~p", [self()]),
    NewState = State0,
    {ok, NewState}.

websocket_handle({text, PlainRequest}, #{counter := Counter0} = State) ->
    ?LOG_DEBUG("[HIGGS-CONN] Receive New Message: ~p, Pid: ~p" , [PlainRequest, self()]),
    NewCounter = Counter0 + 1,
    Reply = list_to_binary("Counter is: " ++ integer_to_list(NewCounter)),
    NewState = #{counter => NewCounter},
    {reply, {text, Reply}, NewState};
    %{ok, State};
websocket_handle(Frame, State) ->
    ?LOG_INFO("[HIGGS-CONN] Invalid Frame => ~p", [Frame]),
    {stop, State}.

websocket_info(Message, State) ->
    ?LOG_INFO("[CONN-INFO] Unhandled message! => Message: ~p", [Message]),
    {reply, {text, list_to_binary(Message)}, State}.
    %{ok, State}.

terminate(Reason, _, State) ->
    ?LOG_INFO("[CONN-TERMINATE] Terminated! => Pid: ~p, Reason: ~p, State: ~p", [self(), Reason, State]),
    ok.

实现思路 #

每个Cowboy WebSocket连接对应一个独立的Erlang进程，核心是通过**OTP内置进程组pg**实现连接的自动存储与索引：

• 利用pg的自动进程清理能力，不需要手动维护连接列表，进程异常退出时会自动从组中移除，避免存储无效Pid
• 收到客户端消息时，拉取进程组内所有连接Pid，排除消息来源后逐个投递转发消息
• 在连接进程的websocket_info回调中处理收到的转发消息，主动推送给绑定的WebSocket客户端

代码改造步骤 #

1. 前置准备 #

在应用启动阶段初始化pg作用域，在应用启动回调中新增：

% 启动时创建聊天室专属的pg作用域，仅执行一次
pg:start(chat_room),

注：pg模块在OTP 23+版本默认自带，低版本Erlang可替换为pg2或ETS表实现。

2. 连接自动注册 #

修改websocket_init回调，新连接建立时自动加入进程组：

websocket_init(State0) ->
    % 将当前连接进程加入名为general的公共聊天室组
    pg:join(chat_room, general, self()),
    ?LOG_DEBUG("[CHAT-CONN] WebSocket连接已建立，pid: ~p", [self()]),
    {ok, State0}.

3. 消息广播逻辑 #

修改websocket_handle中处理文本消息的分支，收到客户端消息后执行全量广播：

websocket_handle({text, PlainRequest}, State) ->
    ?LOG_DEBUG("[CHAT-CONN] 收到客户端消息: ~p, 来源pid: ~p", [PlainRequest, self()]),
    % 拉取所有已接入的聊天室连接Pid列表
    Members = pg:get_members(chat_room, general),
    % 遍历目标连接，排除消息来源后投递广播消息
    lists:foreach(fun(TargetPid) ->
        case TargetPid =:= self() of
            true -> ok;
            false -> TargetPid ! {broadcast, PlainRequest}
        end
    end, Members),
    {ok, State};

4. 转发消息的推送处理 #

修改websocket_info回调，处理其他连接投递的广播消息，推送给当前客户端：

websocket_info({broadcast, Content}, State) ->
    % 收到广播消息时，直接转为WebSocket文本帧推送给客户端
    {reply, {text, Content}, State};
websocket_info(Message, State) ->
    ?LOG_INFO("[CHAT-CONN] 未处理的进程消息: ~p", [Message]),
    {ok, State}.

方案优势 #

• 无额外维护成本：pg自动清理异常退出的连接，不需要在terminate回调中手动删除存储的连接信息
• 并发安全：pg原生支持分布式环境，后续扩展多节点部署时不需要修改核心逻辑
• 低延迟：进程间直接发消息，没有中间队列的额外开销

内容的提问来源于stack exchange，提问作者Sajad Abdollahi