基于core.async实现动态主题订阅的Clojure技术问询

这是个非常典型的 core.async 动态发布/订阅场景，我来分享几个惯用的简洁解决方案，比手动用 ref 维护映射要更可靠且符合 Clojure 风格：

你的需求核心是两个点：

• 动态为新出现的 :sender 自动创建专属订阅者，保证同一发送者的消息串行处理
• 限制全局活跃的消息处理资源（线程/协程数量）

基于 core.async 的 pub 机制，我们可以结合原子状态管理 + 共享工作池来实现，既利用 pub 的主题隔离特性，又避免重复创建订阅者，同时控制资源占用。

1. 基础结构：发布器 + 共享工作池 #

首先搭建全局的发布通道和共享工作池，后者用来限制同时处理消息的活跃资源数：

(require '[clojure.core.async :as async :refer [chan pub sub unsub thread go-loop]])

;; 全局输入通道与发布器，按 :sender 字段拆分主题
(def in-chan (chan))
(def publication (pub in-chan :sender))

;; 共享工作池：限制最大同时处理的任务数为 5（可根据资源调整）
(def max-active-workers 5)
(def worker-pool (chan (async/bounded max-active-workers)))

;; 启动工作池 Worker，负责实际处理消息
(doseq [_ (range max-active-workers)]
  (thread
    (go-loop []
      (when-let [msg (async/<! worker-pool)]
        ;; 这里替换为你的实际消息处理逻辑
        (println (str "Processing message from " (:sender msg) ": " msg))
        (recur)))))

2. 动态订阅管理：原子状态避免重复创建 #

用 atom 维护已订阅的 :sender 集合，保证同一主题不会重复创建订阅者，同时原子性地处理并发场景：

;; 记录已创建订阅者的 sender 集合，避免重复创建
(def subscribed-senders (atom #{}))

;; 确保指定 sender 已有订阅者的辅助函数
(defn ensure-subscribed! [sender]
  (swap! subscribed-senders
         (fn [current-senders]
           (if (contains? current-senders sender)
             current-senders
             ;; 为新 sender 创建订阅通道并绑定到发布器
             (let [sub-chan (chan)]
               (sub publication sender sub-chan)
               ;; 启动串行转发逻辑：保证同一 sender 的消息按顺序进入工作池
               (go-loop []
                 (when-let [msg (async/<! sub-chan)]
                   (async/>! worker-pool msg)
                   (recur)))
               (conj current-senders sender))))))

3. 对外发送接口：自动处理订阅 + 发送消息 #

最后封装对外的发送函数，调用方无需关心订阅逻辑，直接传消息即可：

(defn send-message! [msg]
  (let [sender (:sender msg)]
    (ensure-subscribed! sender)
    ;; 发送消息到发布通道，pub 会自动路由到对应主题的订阅者
    (async/>!! in-chan msg)))

1. 自动订阅的可靠性：

   • swap! 是原子操作，避免了并发下多个线程同时为同一 sender 创建订阅者的问题
   • core.async 的 pub 会自动缓存未订阅主题的消息，直到该主题有订阅者，所以即使发送消息时订阅者还在创建中，消息也不会丢失

2. 消息顺序保证：

   • 每个 sender 的订阅通道是串行读取的（go-loop 单协程处理），所以消息会按到达顺序转发到工作池
   • 工作池的缓冲通道是先进先出，即使多个 Worker 并行处理，同一 sender 的消息也会按顺序被处理

3. 资源控制：

   • 共享工作池的大小直接限制了同时活跃的消息处理线程数，避免无限制创建线程占用资源
   • 如果偏好使用 core.async 的协程池而非原生线程，可以把 thread 换成 go，但注意协程池默认大小是 CPU 核心数 × 2，自定义大小需要修改系统属性

如果订阅逻辑比较重（比如需要初始化额外资源），可以用 Agent 替代 atom，让订阅操作异步串行执行，不阻塞发送消息的线程：

(def subscription-manager (agent #{}))

(defn ensure-subscribed-agent! [sender]
  (send subscription-manager
        (fn [current-senders]
          (if (contains? current-senders sender)
            current-senders
            (let [sub-chan (chan)]
              (sub publication sender sub-chan)
              (go-loop []
                (when-let [msg (async/<! sub-chan)]
                  (async/>! worker-pool msg)
                  (recur)))
              (conj current-senders sender))))))

(defn send-message-agent! [msg]
  (let [sender (:sender msg)]
    (ensure-subscribed-agent! sender)
    (async/>!! in-chan msg)))

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