Clojure 并发设计模式：STM、core.async 与 Agent 实战

(def current-user (atom {:name "Alice" :role "user"}))

;; swap!：原子性更新函数
(swap! current-user assoc :role "admin")
;; => {:name "Alice", :role "admin"}（注意：返回的是新值，不改旧值）

;; 旧值仍然安全可用（不可变保证）

(def account-a (ref 1000))
(def account-b (ref 500))

;; 事务性转账
defn transfer [from to amount]
  (dosync
    (let [from-balance @from]
      (when (< from-balance amount)
        (throw (ex-info "Insufficient funds" {:balance from-balance}))))
    (alter from - amount)
    (alter to + amount)))

(transfer account-a account-b 200)
@account-a  ; => 800
@account-b  ; => 700

(def click-counter (ref 0))

;; commute 更高效（加法可交换，重试不影响最终结果）
(dosync
  (commute click-counter inc))

;; alter 在需要读取检查结果时使用
(defn withdraw [account amount]
  (dosync
    (let [balance @account]
      (if (>= balance amount)
        (alter account - amount)
        (throw (ex-info "余额不足" {}))))))

(def inventory (ref {"apple" 10 "banana" 5}))
(def total-sales (ref 0))

(defn sell-item [item price]
  (dosync
    (let [current-stock ((ensure inventory) item)]
      ;; ensure：标记读取，防止写偏（write skew）
      (when (> current-stock 0)
        (alter inventory update item dec)
        ;; 嵌套事务自动合并到外层
        (alter total-sales + price)))))

;; 并发销售测试
(dotimes [_ 100]
  (future (sell-item "apple" 2)))

(Thread/sleep 500)
@inventory       ; => {"apple" -, "banana" 5}
@total-sales     ; => 正确总计（所有事务或全提交或全回滚）

(def logger-agent (agent []))

;; send：使用线程池（适合 CPU 密集型更新）
(send logger-agent conj "用户登录: Alice")
(send logger-agent conj "订单创建: #12345")

;; send-off：创建新线程（适合 IO 密集型）
(send-off logger-agent
  (fn [logs]
    (spit "app.log" (clojure.string/join "\n" logs))
    []))

;; 等待处理完成
(await logger-agent)
@logger-agent   ; => []（已写入文件后清空）

(def risky-agent (agent 0))

;; 触发错误
(send risky-agent (fn [_] (throw (Exception. "模拟错误"))))

;; 检查错误状态
(agent-error risky-agent)
;; => #object[java.lang.Exception]

;; 重启 agent（清空错误状态）
(restart-agent risky-agent 0 :clear-actions true)

;; 自定义错误处理
(set-error-handler! risky-agent
  (fn [agent ex]
    (println "Agent 错误:" (.getMessage ex))))

(def order-count (agent 0))
(def order-total (ref 0))

(defn process-order [amount]
  (dosync
    (alter order-total + amount))
  ;; 在事务外更新 agent（避免死锁）
  (send order-count inc))

;; Agent 更新函数内可以安全读取 STM 状态
(send order-count
  (fn [n]
    (println (format "已处理 %d 单，总金额 %d" n @order-total))
    n))

(require '[clojure.core.async :as async :refer [<! >! chan go <!! >!! close! go-loop]])

;; 创建缓冲通道
(def c (chan 10))  ; 缓冲容量 10

;; 同步阻塞写入/读取（<!!/>>!! 为主线程阻塞）
(>!! c "hello")    ; 阻塞写入
(<!! c)            ; 阻塞读取 => "hello"

;; 异步 go 块（非阻塞，内部 park）
(go
  (>! c "来自 go block"))

(go
  (println "收到:" (<! c)))

;; alts!：等待多个通道中任意一个就绪
(def ch1 (chan))
(def ch2 (chan))

(go
  (let [[val port] (async/alts! [ch1 ch2])]
    (println "从" port "收到:" val)))

;; 超时机制
(go
  (let [[val port] (async/alts! [ch1 (async/timeout 5000)])]
    (if (= port ch1)
      (println "收到数据:" val)
      (println "超时了！"))))

(defn worker [input-ch output-ch]
  (go-loop []
    (when-let [job (<! input-ch)]
      (let [result (process-job job)]
        (>! output-ch result))
      (recur))))

;; 优雅关闭
(defn shutdown-worker [worker-ch]
  (close! worker-ch))

(defn producer-consumer-system [n-consumers]
  (let [jobs (chan 100)
        results (chan)]
    ;; 启动消费者
    (dotimes [i n-consumers]
      (go-loop [processed 0]
        (if-let [job (<! jobs)]
          (do
            (Thread/sleep (rand-int 100))  ; 模拟工作
            (>! results {:consumer i :job job :processed (inc processed)})
            (recur (inc processed)))
          (println (format "消费者 #%d 处理完成，共 %d 条" i processed)))))
    {:jobs jobs :results results}))

(def system (producer-consumer-system 4))

;; 投递任务
(dotimes [i 20]
  (>!! (:jobs system) i))

;; 收集结果
(go-loop [count 0]
  (when (< count 20)
    (let [result (<! (:results system))]
      (println "完成:" result)
      (recur (inc count)))))

;; 关闭系统
(close! (:jobs system))

(defn pipeline-stage [input-ch output-ch f parallelism]
  (async/pipeline
    parallelism      ; 并行度
    output-ch        ; 输出通道
    (map f)          ; 转换函数
    input-ch         ; 输入通道
    true             ; 关闭输出当输入关闭
    (fn [err] (println "Pipeline 错误:" err)))) ; 错误处理

;; 构建三阶段流水线
(def raw-data (chan 100))
(def parsed-data (chan 100))
(def enriched-data (chan 100))

;; 阶段1：解析
(pipeline-stage raw-data parsed-data parse-json 4)

;; 阶段2：数据增强
(pipeline-stage parsed-data enriched-data enrich-record 8)

;; 阶段3：消费
(go-loop []
  (when-let [record (<! enriched-data)]
    (save-to-db record)
    (recur)))

;; 阻塞通道（容量 0）天然实现背压
(def blocking-ch (chan))

;; 或使用 sliding/dropping 缓冲策略处理溢出
(defn create-safe-producer [buffer-size]
  (chan (async/dropping-buffer buffer-size)))  ; 满时丢弃最旧数据

;; 更精细的背压：基于通道状态调整速率
(defn adaptive-producer [output-ch]
  (go-loop [rate 100]
    (>! output-ch (generate-data))
    ;; 根据通道填充度调整速率
    (let [new-rate (if (> (count output-ch) 80) (/ rate 2) (min (* rate 1.1) 1000))]
      (Thread/sleep (long (/ 1000 rate)))
      (recur new-rate))))

;; Atom：独立变量的同步更新
(def counter (atom 0))
(swap! counter inc)

;; Ref：事务内协调更新
(dosync
  (alter balance-a - amount)
  (alter balance-b + amount))

;; Agent：异步队列更新
(send logger conj log-entry)

;; Channel：数据传递而非共享
(>! channel data)   ; 发送
(<! channel)        ; 接收