原子类型Atom #

一、Atom简介 #

Atom是Clojure中最简单的并发引用类型，用于管理独立的、同步的状态变更。Atom提供无锁的原子操作。

1.1 特点 #

独立状态管理
同步操作
无锁实现（CAS）
原子性保证

1.2 创建Atom #

clojure

(def counter (atom 0))

(def state (atom {:users [] :settings {}}))

@counter

(deref counter)

二、基本操作 #

2.1 读取值 #

clojure

(def counter (atom 0))

@counter

(deref counter)

(deref counter 1000 :timeout)

2.2 reset! #

直接设置新值：

clojure

(def counter (atom 0))

(reset! counter 10)

@counter

(reset! counter {:a 1 :b 2})

@counter

2.3 swap! #

基于当前值计算新值：

clojure

(def counter (atom 0))

(swap! counter inc)

@counter

(swap! counter + 5)

@counter

(swap! counter (fn [x] (* x 2)))

@counter

2.4 swap!重试机制 #

swap! 使用CAS（Compare-And-Swap），如果值被其他线程修改会重试：

clojure

(def counter (atom 0))

(dotimes [_ 1000]
  (future (swap! counter inc)))

(Thread/sleep 100)

@counter

三、比较器操作 #

3.1 compare-and-set! #

clojure

(def counter (atom 0))

(compare-and-set! counter 0 10)

@counter

(compare-and-set! counter 0 20)

@counter

(compare-and-set! counter 10 20)

@counter

3.2 使用场景 #

clojure

(defn try-update [atom old-val new-val]
  (if (compare-and-set! atom old-val new-val)
    :success
    :retry))

(def state (atom :idle))

(try-update state :idle :processing)

(try-update state :idle :done)

四、状态转换 #

4.1 更新嵌套数据 #

clojure

(def app-state (atom {:users []
                      :settings {:theme :light}}))

(swap! app-state update :users conj {:id 1 :name "Alice"})

(swap! app-state update-in [:settings :theme] (constantly :dark))

@app-state

4.2 条件更新 #

clojure

(defn safe-update [atom f]
  (swap! atom
         (fn [current]
           (if (valid? current)
             (f current)
             current))))

(defn valid? [state]
  (and (map? state)
       (contains? state :status)))

(def state (atom {:status :ok}))

(swap! state assoc :count 1)

@state

4.3 事务性更新 #

clojure

(defn update-user [atom user-id f]
  (swap! atom
         (fn [state]
           (if-let [user (get-in state [:users user-id])]
             (assoc-in state [:users user-id] (f user))
             state))))

(def db (atom {:users {1 {:name "Alice" :score 0}
                       2 {:name "Bob" :score 0}}}))

(update-user db 1 update :score inc)

@db

五、观察者模式 #

5.1 add-watch #

clojure

(def counter (atom 0))

(add-watch counter :logger
  (fn [key atom old-val new-val]
    (println "Counter changed from" old-val "to" new-val)))

(swap! counter inc)

(swap! counter + 5)

5.2 remove-watch #

clojure

(remove-watch counter :logger)

(swap! counter inc)

5.3 验证器 #

clojure

(defn positive-validator [value]
  (when (neg? value)
    (throw (ex-info "Value must be positive" {:value value}))))

(def counter (atom 0 :validator positive-validator))

(swap! counter inc)

@counter

(swap! counter dec)

(swap! counter (fn [_] -1))

六、实践示例 #

6.1 计数器 #

clojure

(defn make-counter []
  (atom 0))

(defn increment [counter]
  (swap! counter inc))

(defn decrement [counter]
  (swap! counter dec))

(defn get-count [counter]
  @counter)

(def counter (make-counter))
(increment counter)
(increment counter)
(get-count counter)

6.2 缓存 #

clojure

(defn make-cache []
  (atom {}))

(defn cache-get [cache key]
  (@cache key))

(defn cache-put [cache key value]
  (swap! cache assoc key value))

(defn cache-or-compute [cache key compute-fn]
  (if-let [cached (@cache key)]
    cached
    (let [value (compute-fn)]
      (swap! cache assoc key value)
      value)))

(def my-cache (make-cache))
(cache-or-compute my-cache :expensive #(do (Thread/sleep 100) :result))
(cache-get my-cache :expensive)

6.3 连接池 #

clojure

(defn make-pool [create-fn max-size]
  (atom {:connections []
         :create-fn create-fn
         :max-size max-size}))

(defn borrow-connection [pool]
  (swap! pool
         (fn [{:keys [connections create-fn max-size] :as state}]
           (if (seq connections)
             (-> state
                 (assoc :connections (rest connections))
                 (assoc :borrowed (first connections)))
             state))))

(defn return-connection [pool conn]
  (swap! pool
         (fn [{:keys [connections max-size] :as state}]
           (if (< (count connections) max-size)
             (-> state
                 (assoc :connections (conj connections conn))
                 (dissoc :borrowed))
             state))))

(def pool (make-pool #(str "conn-" (rand-int 1000)) 5))

6.4 状态机 #

clojure

(defn make-state-machine [initial-state transitions]
  (atom {:state initial-state
         :transitions transitions}))

(defn transition [sm event]
  (swap! sm
         (fn [{:keys [state transitions] :as current}]
           (if-let [new-state (get-in transitions [state event])]
             (assoc current :state new-state)
             current))))

(defn current-state [sm]
  (:state @sm))

(def door-sm
  (make-state-machine
    :closed
    {:closed {:open :open}
     :open {:close :closed}}))

(current-state door-sm)
(transition door-sm :open)
(current-state door-sm)
(transition door-sm :close)
(current-state door-sm)

七、性能考虑 #

7.1 swap!重试 #

clojure

(def counter (atom 0))

(time
  (dotimes [_ 10000]
    (swap! counter inc)))

高竞争时重试次数增加，考虑使用更细粒度的状态。

7.2 避免长时间操作 #

clojure

(def state (atom {}))

(swap! state
       (fn [s]
         (Thread/sleep 100)
         (assoc s :key :value)))

swap!中的函数应该快速执行。

八、Atom vs 其他引用类型 #

类型	同步/异步	协调/独立	适用场景
Atom	同步	独立	简单状态
Ref	同步	协调	多变量事务
Agent	异步	独立	后台任务

九、总结 #

Atom操作速查：

操作	说明
`atom`	创建Atom
`deref` / `@`	读取值
`reset!`	设置新值
`swap!`	基于当前值更新
`compare-and-set!`	条件设置
`add-watch`	添加观察者
`remove-watch`	移除观察者

关键点：

Atom用于独立状态的同步管理
swap! 提供原子性保证
高竞争时考虑重试开销
使用观察者模式响应变化
验证器确保状态有效性

下一步，让我们学习引用类型Ref！