基于Pedestal为每个用户实现独立SSE通道的技术咨询

问题背景 #

我正在开发一款曾使用WebSocket的应用，因扩展问题决定替换为Server-Sent Events（SSE）。目标是为每个成功登录的用户创建独立SSE通道，在用户登出或会话超时后关闭通道，避免未使用的通道占用资源。这些用户通道将存储在原子映射注册表中，通过会话token（作为注册表的键）访问特定用户的通道并向前端发送消息。

使用io.pedestal.http.sse命名空间时遇到的问题：该命名空间要求处理器接收一个“默认”通道作为路由处理器的输入参数。若尝试使用独立的用户通道，前端无法接收消息（例如Postman返回空响应），只有依赖处理器输入的“默认”通道时，响应才会包含预期消息。

现有代码片段：

路由定义 #

(def sse-route ["/rest/sse" :get (sse/start-event-stream sse-handler)
    :route-name :sse])

处理器代码 #

(defn sse-handler [channel ctx]
  (doseq [[id {:keys [name data] :as payload}] @LIVE-NOTIFICATIONS]
    (when-not (chan/closed? channel)
      (async/>!! channel payload)
      (swap! LIVE-NOTIFICATIONS dissoc id)
      (Thread/sleep 1000)))
  (async/close! channel))

核心问题：

1. 是否可以使用独立用户通道而非处理器输入的默认通道实现该场景？如何实现？
2. 是否必须自定义拦截器替代start-event-stream，还是仍可使用io.pedestal.http.sse命名空间？
3. 若必须使用输入参数通道，能否在用户登出时访问并关闭它？

解决方案 #

一、可以使用独立用户通道，无需抛弃io.pedestal.http.sse #

你不需要替换start-event-stream，只需将处理器输入的默认通道绑定到用户会话token，存入原子注册表即可实现用户专属通道。具体步骤：

1. 修改处理器，完成通道注册与自动清理
   从请求上下文提取用户会话token，将通道存入注册表，并监听通道关闭事件自动移除条目（防止内存泄漏）：

   ;; 全局原子：存储会话token到SSE通道的映射
   (def user-sse-channels (atom {}))
   
   (defn sse-handler [channel ctx]
     (let [session-token (get-in ctx [:session :token])] ; 根据你的认证逻辑调整token获取方式
       (when session-token
         ;; 注册用户通道
         (swap! user-sse-channels assoc session-token channel)
         ;; 监听通道关闭，自动从注册表移除
         (async/go
           (async/<! channel)
           (swap! user-sse-channels dissoc session-token)))
       ;; 消息发送逻辑：可根据需求改为从业务消息源推送
       (doseq [[id {:keys [name data] :as payload}] @LIVE-NOTIFICATIONS]
         (when-not (chan/closed? channel)
           (async/go (async/>! channel payload)) ; 改用异步发送避免阻塞
           (swap! LIVE-NOTIFICATIONS dissoc id)))
       (async/close! channel)))
   

2. 通过会话token向指定用户推送消息
   直接从注册表中取出对应通道发送消息：

   (defn send-to-user [session-token payload]
     (when-let [channel (get @user-sse-channels session-token)]
       (when-not (chan/closed? channel)
         (async/go (async/>! channel payload)))))
   

二、无需自定义拦截器替代start-event-stream #

start-event-stream已经封装了SSE响应头部设置、通道生命周期管理等底层逻辑，完全可以基于它实现用户专属通道需求，无需重新实现拦截器。

三、登出时可访问并关闭输入参数通道 #

用户登出时，通过会话token从注册表取出通道，关闭并移除即可：

(defn logout-handler [ctx]
  (let [session-token (get-in ctx [:session :token])]
    (when-let [channel (get @user-sse-channels session-token)]
      (when-not (chan/closed? channel)
        (async/close! channel))
      (swap! user-sse-channels dissoc session-token))
    ;; 补充其他登出逻辑（如销毁会话）
    {:status 200 :body "Logged out successfully"}))

同时建议在会话超时处理逻辑中加入相同的通道清理操作，避免残留无效通道。

额外注意事项 #

• 避免阻塞操作：原代码中的Thread/sleep和async/>!!（阻塞发送）会影响性能，建议改用async/go+async/>!的异步发送方式。
• 异常处理：发送消息时需判断通道状态，避免因通道已关闭抛出异常。
• 内存泄漏防护：除了登出主动清理，一定要监听通道关闭事件自动移除注册表条目，覆盖前端主动断开等场景。

内容的提问来源于stack exchange，提问作者Rosen Hristov