Clojure core.async 深入：通道、背压、异常与并发编排

(require '[clojure.core.async :as async])

;; 通道三要素：放（>!）、取（<!）、关闭（close!）
(def ch (async/chan 10))

(async/put! ch :hello)          ;; 非阻塞放（有缓冲）
(async/take! ch (fn [v] ...))   ;; 非阻塞取（回调）

;; go 块内用阻塞式：
(async/go (async/>! ch :x))     ;; go 块内放
(async/go (println (async/<! ch)))  ;; go 块内取

;; 背压的天然形成：
;; 生产者快、消费者慢 → 缓冲满 → 生产者 >! 阻塞
;; → 生产速率被消费者「拖住」（backpressure）
;; 这是特性：无界缓冲才是隐患

;; go-block 模型
;; 1. 遇到 <! 阻塞 → 状态机挂起，让出线程
;; 2. 通道有值 → 状态机恢复，在池中任一线程继续
;; 3. 大量 go 块共享「池」线程，而非每块一线程

(async/go (async/<! ch))   ;; 挂起时不吃线程，代价低

;; 错误：阻塞式调用占住 go 池线程
(async/go (Thread/sleep 1000) ...)   ;; 占线程，不好

;; 正确：真正阻塞的操作放到 thread（独立线程）
(async/thread (Thread/sleep 1000) ...)
;; 或：阻塞操作用 thread + channel 桥接
(def ch (async/to-chan (async/thread (blocking-call!))))

;; 隐患：无界通道 (chan (Integer/MAX_VALUE))
;; 生产者永不阻塞 → 内存无限增长 → OOM
;; 背压 = 有界缓冲 + 阻塞放，二者缺一不可

;; 等多个数据源，谁先到处理谁（扇入）
(defn fan-in [sources]
  (async/go-loop []
    (let [[val ch] (async/alts! sources)]
      (when val
        (process! val)
        (recur)))))

;; 带超时的等（关键模式）
(let [[val ch] (async/alts! [data-ch (async/timeout 5000)])]
  (if (= ch data-ch)
    (handle val)
    (handle-timeout!)))   ;; 超时分支

;; 优先级：请求通道 vs 控制通道
;; 先检查控制通道（如关闭信号），再处理数据
(let [[v c] (async/alts! [ctl-ch req-ch] :priority true)]
  ...)

;; 通道内异常：不会自动传给消费者
(async/go (throw (ex-info "boom" {:x 1})))
;; 消费者永远等不到值（挂死）或拿到 nil（关闭）

;; 显式传播模式：错误「值」化
;; 约定：通道里传结果对象，含 :ok/:err 分支
(async/go (>! ch {:status :ok   :data result}))
(async/go (>! ch {:status :err  :error ex}))

;; 统一「结果通道」约定
(defn wrap-result [f]
  (fn [& args]
    (async/go
      (try {:status :ok :data (apply f args)}
           (catch Exception e
             {:status :err :error e})))))

;; 消费者侧
(async/go
  (let [{:keys [status data error]} (<! ch)]
    (if (= status :ok) (handle data) (log-error! error))))

;; timeout 返回一个「到点自动关闭」的通道
(let [t (async/timeout 1000)]
  (async/go (if (<! t) ...)))   ;; 1s 后 t 关闭 → <! 返回 nil

;; 常见用途：请求超时、心跳、限时等待

;; 多生产者汇合：计数关闭
(defn merge-with-count [chs]
  (let [out (async/chan)]
    (async/go-loop [n (count chs)]
      (if (pos? n)
        (let [[v _] (async/alts! chs)]
          (async/>! out v)
          (recur (if (nil? v) (dec n) n))))   ;; 每个源关闭 -1
      (async/close! out))))

;; core.async 内建 pipeline：从 in 通道读，并行处理，写 out 通道
(defn parallel-pipeline [in out n xf]
  (async/pipeline n out xf in))

;; n = 并行度；xf = 转换函数（如 map）
;; 背压自动：out 满 → pipeline 停读 in

;; 手写多级流水线：采集 → 清洗 → 落库
(def stage1 (async/chan 100))    ;; 原始数据
(def stage2 (async/chan 100))    ;; 清洗后
(def stage3 (async/chan 100))    ;; 待落库

(async/pipeline 4 stage2 (map clean) stage1)
(async/pipeline 2 stage3 (map persist) stage2)
;; stage3 的消费者慢 → 逐级背压，不丢数据

;; 通道传的是不可变值 → 无需锁
;; 唯一可变点是通道本身（线程安全）
;; 状态更新用 atom/ref 或「通道串行化」

(defn fetch-any-with-timeout [fns timeout-ms]
  (let [out  (async/chan 1)
        n    (count fns)]
    ;; 每个任务一个 go：成功 → 写结果并 close，失败 → 记录继续等
    (doseq [f fns]
      (async/go
        (try
          (async/>! out (async/<! (f)))   ;; 任一成功
          (catch Exception _ nil))))
    ;; 主控制：要么拿到结果，要么超时
    (async/<!!
      (async/go
        (let [[v _] (async/alts! [out (async/timeout timeout-ms)])]
          (or v (throw (ex-info "timeout" {:ms timeout-ms}))))))))

;; 用法：三路查询，1s 内任一返回即用，超时抛错
(fetch-any-with-timeout [query-a query-b query-c] 1000)